A2A + MCP + LangChain = 强大的多智能体聊天机器人

结合谷歌的代理到代理协议（A2A）、模型上下文协议（MCP）和LangChain，构建一个能够进行实时股票分析和新闻聚合的自主协作聊天机器人。你已经构建了一个完全集成的多智能体聊天机器人，它利用A2A进行协作，利用MCP进行工具访问，并利用LangChain进行编排。这种架构是自主AI系统的蓝图，这些系统能够思考、适应并协同工作——不再是简单的脚本，而是协作的数字团队。通过将A2A的协作层与MCP

安思派Anspire

812人浏览 · 2025-11-14 10:50:35

安思派Anspire · 2025-11-14 10:50:35 发布

结合谷歌的代理到代理协议（A2A）、模型上下文协议（MCP）和LangChain，构建一个能够进行实时股票分析和新闻聚合的自主协作聊天机器人。在本深度指南中，我们将逐步介绍如何设置服务器、编写工具，并将所有内容整合到一个单一的元代理中。

按回车键或点击以查看全尺寸图像

为什么A2A和MCP很重要

智能体到智能体协议（A2A）：使AI智能体能够发现、通信并协同解决问题——这与专家在团队中协作的方式类似。
模型上下文协议（MCP）：定义了语言模型访问外部工具和数据源的标准接口，类似于在代码中调用 API。

通过将A2A的协作层与MCP的工具访问层相结合，您可以从硬编码脚本过渡到编排多个AI组件的动态、自主系统。

实时演示：运行中的聊天机器人

场景：你提问：

“苹果和NVIDIA目前的股价是多少，有什么最新消息？”

在幕后

元代理路由：中央控制器会检查您的查询，选择要调用的子代理和工具。
StockData工具：解析股票代码（AAPL、NVDA），通过yfinance获取价格，计算诸如百分比变化和市值等指标。
财经新闻工具：从Finviz抓取头条新闻，解析相对URL，收集公司概况。
A2A专家代理：使用OpenAI的GPT模型提供专业分析。
聚合：将原始数据、抓取的新闻和专家评论整合为统一的、易于理解的响应。

先决条件

在你的Python环境中安装所有必需的库：

pip install python-a2a fastmcp langchain-openai yfinance pandas requests beautifulsoup4

确保你已经设置了环境变量OPENAI_API_KEY，并将其值设置为你的 OpenAI 密钥。

1. 设置A2A服务器

将由OpenAI驱动的金融专家包装成A2A代理：

import os
from python_a2a import OpenAIA2AServer, run_server
from python_a2a import AgentCard, AgentSkill
# Define the agent's profile
agent_card = AgentCard(
    name="Stock Market Expert",
    description="Expert in market trends, fundamentals, and investment strategies.",
    url="http://localhost:5000",
    version="1.0.0",
    skills=[
        AgentSkill(
            name="Market Analysis",
            description="Analyze overall market sentiment and key indicators.",
            examples=["What's the current market sentiment?", "Impact of interest rates on tech stocks?"]
        ),
        AgentSkill(
            name="Investment Strategies",
            description="Discuss risk management and portfolio diversification.",
            examples=["How to diversify my portfolio?", "Explanation of dollar-cost averaging."]
        ),
        AgentSkill(
            name="Company Analysis",
            description="Interpret financial ratios and company fundamentals.",
            examples=["How to read P/E ratios?", "Key metrics for evaluating growth stocks."]
        )
    ]
)
# Initialize and start the A2A server
a2a_server = OpenAIA2AServer(
    api_key=os.environ['OPENAI_API_KEY'],
    model="gpt-4o",
    temperature=0.0,
    system_prompt="You are a stock market and financial analysis expert. Provide factual, concise insights."
)
if __name__ == 'main':
    run_server(a2a_server, host='0.0.0.0', port=5000)

提示：使用后台线程或进程管理器（例如gunicorn）来保持代理持续运行。

2. 构建MCP工具

初始化MCP服务器并定义两个重要工具。

from python_a2a.mcp import FastMCP
# Create MCP server
mcp_server = FastMCP(
    name="FinanceTools",
    description="Tools for retrieving stock data and financial news."
)

2.1 股票数据获取器

import re
yfinance
import pandas as pd
@mcp_server.tool(
    name="stock_data",
    description="Fetch metrics for stocks by ticker symbols or names."
)
def stock_data(input_str=None, **kwargs):
    input_str = kwargs.get('input', input_str)
    if not input_str:
        return {"error": "No input provided."}
    # Extract tickers
    tickers = []
    if ',' in input_str:
        tickers = [t.strip().upper() for t in input_str.split(',')]
    else:
        tickers = [w.upper() for w in re.findall(r"\b[A-Za-z]{1,5}\b", input_str)]
    # Fallback common names
    common = {'apple':'AAPL', 'nvidia':'NVDA'}
    if not tickers:
        for name,t in common.items():
            if name in input_str.lower(): tickers.append(t)
    results = {}
    for t in tickers:
        tk = yfinance.Ticker(t)
        hist = tk.history(period="1mo")
        if hist.empty:
            results[t] = {"error": "No data."}
            continue
        first, last = hist.iloc[0], hist.iloc[-1]
        change = float(last['Close'] - first['Close'])
        pct = change / float(first['Close']) * 100
        info = tk.info
        summary = {
            "latest_price": float(last['Close']),
            "price_change": change,
            "%_change": pct,
            "52_week_high": info.get('fiftyTwoWeekHigh'),
            "52_week_low": info.get('fiftyTwoWeekLow'),
            "market_cap": info.get('marketCap'),
            "pe_ratio": info.get('trailingPE')
        }
        results[t] = summary
    return results

2.2 财经新闻爬虫

import requests
from bs4 import BeautifulSoup
from urllib.parse import urljoin
@mcp_server.tool(
    name="web_scraper",
    description="Scrape latest headlines and company snapshot from Finviz."
)
def web_scraper(input_str=None, **kwargs):
    ticker = (kwargs.get('input') or input_str or '').upper()
    url = f"https://finviz.com/quote.ashx?t={ticker.lower()}"
    headers = {'User-Agent':'Mozilla/5.0'}
    resp = requests.get(url, headers=headers)
    soup = BeautifulSoup(resp.text, 'html.parser')
    news = []
    for row in soup.select('#news-table tr')[:5]:
        date, title = row.find_all('td')
        link = title.a['href']
        if not link.startswith('http'):
            link = urljoin(url, link)
        news.append({"date": date.text, "title": title.text.strip(), "link": link})
    details = {}
    snap = soup.find('table', {'class':'snapshot-table2'})
    for r in snap.find_all('tr'):
        cells = r.find_all('td')
        for i in range(0, len(cells), 2):
            details[cells[i].text] = cells[i+1].text
    return {"news_items": news, "snapshot": details}

在端口6000上启动MCP服务器：

python -c "from your_module import mcp_server; mcp_server.run(host='0.0.0.0', port=6000)"

3. 与LangChain集成

将A2A和MCP服务器都转换为LangChain工具：

from python_a2a.langchain import to_langchain_agent, to_langchain_tool
from langchain_openai import ChatOpenAI
from langchain.agents import initialize_agent, Tool, AgentType
# A2A → LangChain
a2a_agent = to_langchain_agent("http://localhost:5000")
# MCP → LangChain
stock_tool = to_langchain_tool("http://localhost:6000", "stock_data")
news_tool = to_langchain_tool("http://localhost:6000", "web_scraper")
# Define wrapper functions
def ask_expert(q): return a2a_agent.invoke(q).get('output')
def fetch_data(q): return stock_tool.invoke(q)
def fetch_news(q): return news_tool.invoke(q)
# Assemble Tool objects
tools = [
    Tool(name="StockExpert", func=ask_expert, description="Ask financial questions."),
    Tool(name="StockData", func=fetch_data, description="Retrieve stock metrics."),
    Tool(name="FinancialNews", func=fetch_news, description="Get latest financial headlines.")
]
# Initialize meta-agent
llm = ChatOpenAI(model="gpt-4o", temperature=0.0)
meta_agent = initialize_agent(
    tools, llm, agent=AgentType.OPENAI_FUNCTIONS, verbose=True
)

4. 创建元智能体

在主脚本中将所有内容整合在一起：

if __name__ == '__main__':
    query = "What are the current stock prices of Apple and NVIDIA, plus top news?"
    response = meta_agent.invoke(query)
    print("Meta-Agent Response:\n", response)

这一次调用就能触发动态工具选择、数据检索、新闻抓取和专家评论——一气呵成。