从零构建AI股票分析系统：A2A+MCP+LangChain打造专业投研团队(必收藏)

文章详细介绍了如何结合A2A、MCP和LangChain技术构建股票分析AI系统，包括服务器搭建、工具开发、元代理整合等完整实现流程。系统通过多代理协作实现股票数据获取、新闻抓取和专业分析，最终提供统一报告，同时提供AI大模型学习资料，帮助开发者掌握相关技术。

datian1234

813人浏览 · 2026-02-16 08:45:00

datian1234 · 2026-02-16 08:45:00 发布

在人工智能技术不断发展的今天，构建一个能够自主分析股票行情、抓取财经新闻并提供专业解读的聊天机器人，不再是科幻，而是现实可能。

结合 Google 的代理到代理协议（A2A）、模型上下文协议（MCP）和 LangChain，我们可以打造一个多代理协作、动态响应的智能系统，让 AI 真正像一支专业的投研团队一样工作。

在这份深入指南中，我将带你一步步实现从服务器搭建、工具开发，到元代理整合的全过程，让复杂的多代理系统落地成为可操作的项目。

为什么 A2A 和 MCP 这么关键？Agent-to-Agent Protocol（A2A）让人工智能代理能够发现彼此、互相沟通并协作解决问题，这正如企业中的不同专家协同工作一般。

而 Model Context Protocol（MCP）则为语言模型提供了统一接口，用于访问外部工具和数据源，类似于我们在编程中调用 API。将 A2A 的协作能力与 MCP 的工具访问能力结合，你就可以从固定脚本迈向自主调度、多组件协作的智能系统。

想象一个场景，你向聊天机器人提问：“苹果和英伟达的最新股价是多少？最近有什么新闻？”在后台，元代理会根据问题自动路由，选择合适的子代理和工具。

股票数据工具解析股票代码（AAPL、NVDA），通过 yfinance 获取定价，并计算涨跌幅和市值；财经新闻工具抓取 Finviz 的头条新闻，解析相关 URL 并收集公司快照；A2A 专家代理则使用 GPT 模型提供专业分析。

最终，这些信息被聚合成一份统一的人类可读的报告，实现数据、新闻与智能分析的无缝整合。

在开始之前，确保你的 Python 环境已经安装了必要库：

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

同时，记得将环境变量 OPENAI_API_KEY 设置为你的 OpenAI 密钥。

首先，我们来搭建 A2A 服务器，将 OpenAI 支持的金融专家包装为 A2A 代理：

import osfrom python_a2a import OpenAIA2AServer, run_serverfrom python_a2a import AgentCard, AgentSkillagent_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."]        )    ])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）。

接下来，构建 MCP 工具，初始化 MCP 服务器并定义两个关键工具：

from python_a2a.mcp import FastMCPmcp_server = FastMCP(    name="FinanceTools",    description="Tools for retrieving stock data and financial news.")

股票数据获取器通过解析股票代码获取历史行情并计算指标：

import reimport yfinanceimport 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."}    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)]    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

财经新闻抓取工具通过解析 Finviz 页面获取新闻和公司快照：

import requestsfrom bs4 import BeautifulSoupfrom 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}

启动 MCP 服务器：

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

将 A2A 与 MCP 集成到 LangChain 中，实现统一调用和动态调度：

from python_a2a.langchain import to_langchain_agent, to_langchain_toolfrom langchain_openai import ChatOpenAIfrom langchain.agents import initialize_agent, Tool, AgentTypea2a_agent = to_langchain_agent("http://localhost:5000")stock_tool = to_langchain_tool("http://localhost:6000", "stock_data")news_tool = to_langchain_tool("http://localhost:6000", "web_scraper")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)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.")]llm = ChatOpenAI(model="gpt-4o", temperature=0.0)meta_agent = initialize_agent(    tools, llm, agent=AgentType.OPENAI_FUNCTIONS, verbose=True)

最后，将所有功能绑在主脚本中，一次调用即可触发完整流程：

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)

这一次调用完成了动态工具选择、数据检索、新闻抓取和专业分析的全部流程。为了确保系统稳定，检查服务器状态，验证端口是否占用，并注意 pandas/numpy 类型的 JSON 序列化问题。

使用正则解析和名称到代码映射保证输入灵活性，Try/except 块提供清晰的错误信息。LangChain 的 OPENAI_FUNCTIONS 代理会根据问题自动选择最合适的工具，实现真正的智能协作。

通过这一系统，你已经构建了一个完全集成的多代理聊天机器人，它利用 A2A 协作、MCP 工具访问以及 LangChain 编排，使 AI 不再是孤立脚本，而是一支能够思考、适应、协作的数字团队。

接下来，你可以扩展工具以支持投资组合跟踪、情绪分析或期权数据，并在 Kubernetes 或 Docker 上部署，实现可扩展性；同时，还可以集成 Web UI 或 Slack 接口，让这支数字团队触手可及。