在人工智能技术快速发展的背景下，构建一个能自主分析股票行情、抓取财经新闻并提供专业解读的聊天机器人已从概念变为现实。
通过结合 Google 的代理到代理协议（A2A）、模型上下文协议（MCP）与 LangChain 框架，可打造一个多代理协同、动态响应的智能投研系统，让 AI 像专业团队一样运作。
本文将带你完成服务器搭建、工具开发到元代理整合的全流程，实现多代理系统的工程化落地。
A2A 与 MCP：构建智能协作的核心
Agent-to-Agent Protocol（A2A）使 AI 代理能够发现彼此、通信并协同解决问题，模拟专家团队协作机制。Model Context Protocol（MCP）则为大模型提供统一接口，用于调用外部工具和数据源，类似于 API 调用。
二者结合，可实现从静态脚本向自主调度、多组件联动的智能系统跃迁。
系统工作流程
当用户提问“苹果和英伟达最新股价是多少？有何相关新闻？”时，系统自动触发以下流程：
- 元代理解析问题，路由至相应子代理；
- 股票数据工具通过 yfinance 获取 AAPL 和 NVDA 的实时价格、涨跌幅、市值等指标；
- 财经新闻工具抓取 Finviz 头条资讯，提取标题、链接及发布时间；
- A2A 专家代理调用 GPT 模型进行综合分析；
- 最终输出结构清晰、内容完整的可读报告。
环境准备
确保已安装以下 Python 库：
pip install python-a2a fastmcp langchain-openai yfinance pandas requests beautifulsoup4同时设置环境变量 OPENAI_API_KEY 为你的 OpenAI 密钥。
搭建 A2A 服务器
将基于 OpenAI 的金融专家封装为 A2A 代理：
import os
from python_a2a import OpenAIA2AServer, run_server
from python_a2a import AgentCard, AgentSkill
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."]
)
]
)
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 FastMCP
mcp_server = FastMCP(
name="FinanceTools",
description="Tools for retrieving stock data and financial news."
)
股票数据获取工具
支持通过股票代码或公司名称查询历史行情与核心指标：
import re
import 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."}
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 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):
if i+1 < len(cells):
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)"集成至 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_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)
该调用将自动完成问题解析、工具选择、数据获取、新闻采集与专业分析，最终生成整合报告。
优化建议与扩展方向
为保障系统稳定性，需注意：
- 检查端口占用情况，确保 A2A（5000）与 MCP（6000）服务正常运行；
- 处理 pandas/numpy 数据类型在 JSON 序列化中的兼容性问题；
- 使用 try-except 结构提升错误处理能力；
- 借助正则表达式与名称映射增强输入识别灵活性。
LangChain 的 OPENAI_FUNCTIONS 代理可实现智能工具选择，真正达成自主协作。
未来可扩展方向包括：
- 增加投资组合跟踪、情绪分析、期权数据等功能模块；
- 通过 Docker 或 Kubernetes 部署，提升系统可扩展性；
- 集成 Web 前端或 Slack 插件，提升交互体验。
通过此架构，你已构建出一个集 A2A 协作、MCP 工具调用与 LangChain 编排于一体的多代理智能系统，赋予 AI 类似专业团队的综合决策能力。