A2A + MCP：构建实用人工智能系统的超强组合

构建真正有效的连接型人工智能系统的挑战

如果你正在构建人工智能应用，这种情况可能听起来很熟悉：

你需要特定的人工智能能力来解决业务问题。你找到了完成每个单独任务的出色工具。但把所有东西连接在一起却占据了大部分开发时间，还创建了一个脆弱的系统，一旦有任何更改就会崩溃。

好消息是，有一个今天就能用的实用解决方案。

A2A + MCP：完美组合

这两个协议协同工作，解决了完整的集成挑战：

A2A（智能体到智能体） 就像是智能体的社交网络。它让它们可以直接相互沟通和协作。想象一下，你的日历智能体可以自动与你的旅行智能体协调，当航班延误时重新安排会议。

MCP（模型上下文协议） 更像是一个通用适配器。它为人工智能模型提供了标准化的方式以访问工具和数据源。它使得你的智能助手能够轻松地查看天气或搜索知识库。

A2A 处理智能体之间的通信，而 MCP 将智能体连接到它们的工具。

它们共同创建了一个清晰的双层架构，这将为你节省数百小时的开发时间。

展示如何结合 a2a 和 mcp 的示例

构建你的第一个 MCP 服务器

让我们从实际操作开始：创建一个暴露有用工具的 MCP 服务器。它有多简单，看看就知道了：

from python_a2a.mcp import FastMCP, text_response# 创建一个新的 MCP 服务器
calculator_mcp = FastMCP(name="Calculator MCP",version="1.0.0",description="Provides mathematical calculation functions"
)
# 使用简单的装饰器和类型提示定义工具
@calculator_mcp.tool()
def add(a: float, b: float) -> float:"""将两个数字相加。"""return a + b
@calculator_mcp.tool()
def subtract(a: float, b: float) -> float:"""从 a 中减去 b。"""return a - b
@calculator_mcp.tool()
def multiply(a: float, b: float) -> float:"""将两个数字相乘。"""return a * b
@calculator_mcp.tool()
def divide(a: float, b: float) -> float:"""将 a 除以 b。"""if b == 0:return text_response("Cannot divide by zero")return a / b
# 运行服务器
if __name__ == "__main__":calculator_mcp.run(host="0.0.0.0", port=5001)

就这样——你创建了一个暴露数学运算作为标准化工具的 MCP 服务器。注意类型提示和文档字符串是如何自动成为工具接口的一部分的，这使得它们具有自我说明性。

将 MCP 工具连接到 A2A 智能体

现在，让我们创建一个使用这些 MCP 工具的 A2A 智能体：

from python_a2a import A2AServer, Message, TextContent, MessageRole, run_server
from python_a2a.mcp import FastMCPAgentclass CalculatorAgent(A2AServer, FastMCPAgent):"""使用 MCP 工具的计算器智能体。"""def __init__(self):# 初始化父类A2AServer.__init__(self)FastMCPAgent.__init__(self,mcp_servers={"calc": "http://localhost:5001"}  # 连接到我们的 MCP 服务器)async def handle_message_async(self, message):"""处理消息并使用 MCP 工具。"""try:if message.content.type == "text":text = message.content.text.lower()# 从消息中提取数字import renumbers = [float(n) for n in re.findall(r"[-+]?\d*\.?\d+", text)]if len(numbers) >= 2:# 确定要执行的操作if "add" in text or "plus" in text or "+" in text:# 调用 MCP 工具执行计算result = await self.call_mcp_tool("calc", "add", a=numbers[0], b=numbers[1])return Message(content=TextContent(text=f"The sum of {numbers[0]} and {numbers[1]} is {result}"),role=MessageRole.AGENT,parent_message_id=message.message_id,conversation_id=message.conversation_id)elif "subtract" in text or "minus" in text or "-" in text:result = await self.call_mcp_tool("calc", "subtract", a=numbers[0], b=numbers[1])return Message(content=TextContent(text=f"The difference between {numbers[0]} and {numbers[1]} is {result}"),role=MessageRole.AGENT,parent_message_id=message.message_id,conversation_id=message.conversation_id)# 添加其他操作...# 默认响应return Message(content=TextContent(text="I'm a calculator agent. You can ask me to add, subtract, multiply, or divide numbers."),role=MessageRole.AGENT,parent_message_id=message.message_id,conversation_id=message.conversation_id)except Exception as e:# 处理错误return Message(content=TextContent(text=f"Error: {str(e)}"),role=MessageRole.AGENT,parent_message_id=message.message_id,conversation_id=message.conversation_id)
# 运行智能体
if __name__ == "__main__":agent = CalculatorAgent()run_server(agent, host="0.0.0.0", port=5000)

现在你有了一个使用 MCP 工具的 A2A 智能体。用户可以用自然语言与这个智能体互动，比如问“5 加 3 等于多少？”智能体会调用相应的 MCP 工具来计算答案。

这种做法的美妙之处在于清晰的分离：

• MCP 服务器专注于提供计算能力
• A2A 智能体处理自然语言理解和用户互动
• 它们之间的连接是标准化的且无状态的

完整的实用示例：股票信息查询系统

现在让我们构建一个更有用的东西：一个帮助用户查找股票价格的股票信息查询系统，即使他们不知道股票代码也能查到。

1. 为特定工具创建 MCP 服务器

首先，我们将为两个关键能力创建 MCP 服务器：

# DuckDuckGo MCP 服务器，用于查找股票代码
duckduckgo_mcp = FastMCP(name="DuckDuckGo MCP",version="1.0.0",description="搜索功能，用于查找股票信息"
)@duckduckgo_mcp.tool()
def search_ticker(company_name: str) -> str:"""使用 DuckDuckGo 搜索查找公司的股票代码。"""# 实现细节（简化版）import requestsimport requery = f"{company_name} stock ticker symbol"url = f"https://api.duckduckgo.com/?q={query}&format=json"response = requests.get(url)data = response.json()# 从结果中提取代码# 查找类似“NYSE: AAPL”或“NASDAQ: MSFT”的模式text = data.get("Abstract", "")ticker_match = re.search(r'(?:NYSE|NASDAQ):\s*([A-Z]+)', text)if ticker_match:return ticker_match.group(1)# 为了演示目的，回退到常见代码if company_name.lower() == "apple":return "AAPL"elif company_name.lower() == "microsoft":return "MSFT"# 等等...return f"Could not find ticker for {company_name}"
# 在端口 5001 上运行此服务器
duckduckgo_mcp.run(port=5001)  # 在真实应用中，将其作为单独的进程运行# YFinance MCP 服务器，用于股票价格数据
yfinance_mcp = FastMCP(name="YFinance MCP",version="1.0.0",description="股票市场数据工具"
)@yfinance_mcp.tool()
def get_stock_price(ticker: str) -> dict:"""获取给定股票代码的当前股票价格。"""# 实现细节import yfinance as yftry:# 获取股票数据stock = yf.Ticker(ticker)data = stock.history(period="1d")if data.empty:return {"error": f"No data found for ticker {ticker}"}# 提取价格price = data['Close'].iloc[-1]return {"ticker": ticker,"price": price,"currency": "USD","timestamp": data.index[-1].strftime('%Y-%m-%d %H:%M:%S')}except Exception as e:return {"error": f"Error fetching stock data: {str(e)}"}
# 在端口 5002 上运行此服务器
yfinance_mcp.run(port=5002)  # 在真实应用中，将其作为单独的进程运行

2. 创建使用这些 MCP 工具的 A2A 智能体

现在让我们创建暴露这些能力的 A2A 智能体：

# DuckDuckGo 智能体，用于查找股票代码
class DuckDuckGoAgent(A2AServer, FastMCPAgent):"""查找股票代码的智能体。"""def __init__(self):A2AServer.__init__(self)FastMCPAgent.__init__(self,mcp_servers={"search": "http://localhost:5001"})async def handle_message_async(self, message):if message.content.type == "text":# 从消息中提取公司名称import recompany_match = re.search(r"ticker\s+(?:for|of)\s+([A-Za-z\s]+)", message.content.text, re.I)if company_match:company_name = company_match.group(1).strip()else:# 默认使用整个消息company_name = message.content.text.strip()# 调用 MCP 工具查找代码ticker = await self.call_mcp_tool("search", "search_ticker", company_name=company_name)return Message(content=TextContent(text=f"The ticker symbol for {company_name} is {ticker}."),role=MessageRole.AGENT,parent_message_id=message.message_id,conversation_id=message.conversation_id)# 处理其他消息类型或错误return Message(content=TextContent(text="I can help find ticker symbols for companies."),role=MessageRole.AGENT,parent_message_id=message.message_id,conversation_id=message.conversation_id)
# 在端口 5003 上运行此智能体
run_server(DuckDuckGoAgent(), port=5003)  # 在真实应用中，将其作为单独的进程运行
# YFinance 智能体，用于股票价格
class YFinanceAgent(A2AServer, FastMCPAgent):"""提供股票价格信息的智能体。"""def __init__(self):A2AServer.__init__(self)FastMCPAgent.__init__(self,mcp_servers={"finance": "http://localhost:5002"})async def handle_message_async(self, message):if message.content.type == "text":# 从消息中提取代码import reticker_match = re.search(r"\b([A-Z]{1,5})\b", message.content.text)if ticker_match:ticker = ticker_match.group(1)# 调用 MCP 工具获取价格price_info = await self.call_mcp_tool("finance", "get_stock_price", ticker=ticker)if "error" in price_info:return Message(content=TextContent(text=f"Error getting price for {ticker}: {price_info['error']}"),role=MessageRole.AGENT,parent_message_id=message.message_id,conversation_id=message.conversation_id)return Message(content=TextContent(text=f"{ticker} is currently trading at {price_info['price']:.2f} {price_info['currency']}."),role=MessageRole.AGENT,parent_message_id=message.message_id,conversation_id=message.conversation_id)# 处理其他消息类型或错误return Message(content=TextContent(text="I can provide stock price information for ticker symbols."),role=MessageRole.AGENT,parent_message_id=message.message_id,conversation_id=message.conversation_id)
# 在端口 5004 上运行此智能体
run_server(YFinanceAgent(), port=5004)  # 在真实应用中，将其作为单独的进程运行

3. 创建一个协调器来整合这些能力

最后，让我们创建一个股票助手，整合这些专业智能体，提供无缝体验：

from python_a2a import A2AClient, Message, TextContent, MessageRole, OpenAIA2AServer, run_server
import re
import osclass StockAssistant(OpenAIA2AServer):"""一个股票信息助手，协调专业智能体。"""def __init__(self, api_key, duckduckgo_endpoint, yfinance_endpoint):# 初始化 OpenAI 驱动的智能体super().__init__(api_key=api_key,model="gpt-3.5-turbo",system_prompt=("You are a helpful financial assistant that helps users get stock information. ""You extract company names from user queries to find ticker symbols and prices."))# 创建连接到专业智能体的客户端self.duckduckgo_client = A2AClient(duckduckgo_endpoint)self.yfinance_client = A2AClient(yfinance_endpoint)def handle_message(self, message):"""覆盖以拦截股票相关查询。"""if message.content.type == "text":text = message.content.text.lower()# 检查这是否是股票价格查询if ("stock" in text or "price" in text or "trading" in text) and any(company in text for company in ["apple", "microsoft", "google", "amazon", "tesla", "facebook", "meta"]):# 将其作为股票查询处理return self._get_stock_info(message)# 对于所有其他消息，委托给 OpenAIreturn super().handle_message(message)def _get_stock_info(self, message):"""通过协调专业智能体处理股票信息查询。"""try:# 首先，使用 OpenAI 提取公司名称openai_response = super().handle_message(Message(content=TextContent(text=f"Extract only the company name from this query: '{message.content.text}'. "f"Return ONLY the company name, nothing else."),role=MessageRole.USER))company_name = openai_response.content.text.strip()company_name = company_name.strip('"\'.,')# 第一步：从 DuckDuckGo 智能体获取股票代码ticker_message = Message(content=TextContent(text=f"What's the ticker for {company_name}?"),role=MessageRole.USER)ticker_response = self.duckduckgo_client.send_message(ticker_message)# 从响应中提取代码ticker_match = re.search(r'ticker\s+(?:symbol\s+)?(?:for\s+[\w\s]+\s+)?is\s+([A-Z]{1,5})',ticker_response.content.text, re.I)if not ticker_match:return Message(content=TextContent(text=f"I couldn't find the ticker symbol for {company_name}."),role=MessageRole.AGENT,parent_message_id=message.message_id,conversation_id=message.conversation_id)ticker = ticker_match.group(1)# 第二步：从 YFinance 智能体获取股票价格price_message = Message(content=TextContent(text=f"What's the current price of {ticker}?"),role=MessageRole.USER)price_response = self.yfinance_client.send_message(price_message)# 返回合并后的信息return Message(content=TextContent(text=f"{company_name} ({ticker}): {price_response.content.text}"),role=MessageRole.AGENT,parent_message_id=message.message_id,conversation_id=message.conversation_id)except Exception as e:# 处理任何错误return Message(content=TextContent(text=f"Sorry, I encountered an error: {str(e)}"),role=MessageRole.AGENT,parent_message_id=message.message_id,conversation_id=message.conversation_id)
# 运行助手
if __name__ == "__main__":api_key = os.environ.get("OPENAI_API_KEY")if not api_key:print("Error: OPENAI_API_KEY environment variable is required")exit(1)assistant = StockAssistant(api_key=api_key,duckduckgo_endpoint="http://localhost:5003/a2a",yfinance_endpoint="http://localhost:5004/a2a")run_server(assistant, port=5000)

4. 创建一个简单的客户端供用户使用

最后，让我们创建一个客户端，方便用户与我们的系统互动：

from python_a2a import A2AClient, Message, TextContent, MessageRole
import argparsedef interactive_session(client):"""与股票助手的交互式会话。"""print("\n===== Stock Price Assistant =====")print("Type 'exit' or 'quit' to end the session.")print("Example queries:")print("  - What's the stock price of Apple?")print("  - How much is Microsoft trading for?")print("  - Get the current price of Tesla stock")print("=" * 50)while True:try:user_input = input("\n> ")if user_input.lower() in ["exit", "quit"]:print("Goodbye!")break# 以文本消息发送message = Message(content=TextContent(text=user_input),role=MessageRole.USER)print("Sending to assistant...")response = client.send_message(message)# 显示响应print(f"\nAssistant: {response.content.text}")except Exception as e:print(f"Error: {e}")print("Please try again or type 'exit' to quit.")
if __name__ == "__main__":parser = argparse.ArgumentParser(description="Stock Assistant Client")parser.add_argument("--endpoint", default="http://localhost:5000/a2a",help="Stock assistant endpoint URL")args = parser.parse_args()# 创建客户端client = A2AClient(args.endpoint)# 开始交互式会话interactive_session(client)

这种架构为你创造的价值

这种 A2A + MCP 架构今天就能带来实实在在的实用价值：

1. 大幅减少集成工作：无需为每个服务编写自定义连接器——一切都能说同一种语言。
2. 轻松替换组件：需要更好的数据源？只需插入一个新的智能体——协调器保持不变。
3. 清晰的错误边界：每个智能体处理自己的错误，防止系统性故障。
4. 简单可扩展：想添加历史数据？只需创建一个新的 MCP 工具，并通过新的或现有的智能体暴露它。
5. 跨项目复用组件：那个 DuckDuckGo 智能体可以在任何需要搜索功能的项目中使用。

今天就能运行这个系统

你可以按照以下步骤今天就运行这个完整的股票信息查询系统：

设置你的环境：

# 安装必要的包pip install "python-a2a[all]" yfinance requests

启动每个组件（在单独的终端中）：

# 启动 MCP 服务器
python duckduckgo_mcp_server.py
python yfinance_mcp_server.py# 启动 A2A 智能体
python duckduckgo_agent.py
python yfinance_agent.py# 启动协调器（替换为你的 API 密钥） OPENAI_API_KEY=your_key_here
python stock_assistant.py  # 运行客户端 python stock_client.py

通过客户端与系统互动：

> What's the current stock price of Apple?
Assistant: Apple (AAPL): AAPL is currently trading at 173.58 USD.

你可以立即解决的三个真实业务问题

这种架构模式可以立即解决许多业务问题。以下是三个例子：

1. 客户支持自动化

问题：支持团队需要访问多个后端系统来回答客户问题。

A2A + MCP 解决方案：

• 为每个后端系统创建 MCP 工具（订单、产品、运输）
• 构建每个领域的专业 A2A 智能体
• 创建一个协调器，路由问题并聚合答案

创造的价值：

• 支持代理可以在一个地方获得完整的信息
• 后端系统保持隔离但可访问
• 添加新的数据源无需重新训练整个系统

2. 文档处理流程

问题：处理文档需要多个专业步骤（OCR、提取、分类）。

A2A + MCP 解决方案：

• 为每个处理步骤创建 MCP 工具
• 构建专门处理不同文档类型的 A2A 智能体
• 创建一个协调器来管理工作流

创造的价值：

• 文档处理步骤之间有清晰的分离
• 不同文档类型的专业处理
• 容易添加对新文档格式的支持

3.研究助理：

问题：研究人员需要查询文献、分析数据并应用领域知识。

A2A + MCP 解决方案：

• 创建用于文献搜索、数据分析和领域特定计算的 MCP 工具。
• 构建专门处理不同研究领域的 A2A 智能体。
• 创建一个协调器来处理复杂的查询请求。

创造的价值：

• 研究人员可以从多个来源获得全面的答案。
• 正确应用领域特定的计算。
• 添加新的研究方法不会破坏现有功能。

如何构建你自己的 A2A + MCP 系统

你可以立即开始实现这种架构，只需按照以下步骤操作：

确定你的能力需求：

• 你需要哪些工具或数据源？
• 涉及哪些专业领域知识？

为每种能力创建 MCP 工具：

• 定义清晰的接口，并使用类型提示。
• 使用适当的错误处理。
• 让每个工具专注于一种能力。

构建使用这些工具的 A2A 智能体：

• 每个智能体应专注于一个领域。
• 使用 MCP 来访问工具。
• 通过 A2A 暴露能力。

创建协调各个智能体的协调器：

• 路由查询请求到适当的智能体。
• 将结果合并成连贯的响应。
• 管理对话流程。

构建用户友好的客户端：

• 通过 A2A 连接到协调器。
• 清晰地向用户展示信息。
• 优雅地处理错误。

总结

A2A + MCP 不仅仅是一个理论架构，它是一个今天就可以实现的实用解决方案，用于解决真实业务问题。通过标准化智能体通信和工具集成，你可以构建出：

• 模块化：易于扩展和维护。
• 可靠：清晰的错误边界和稳健的通信。
• 灵活：组件可以独立替换或升级。
• 可复用：工具和智能体可以在多个项目中共享。

别再浪费时间构建自定义集成，A2A + MCP 已经提供了一个经过验证的解决方案。

今天就试试看吧

Python A2A 库现在已经完全支持 MCP。你可以立即开始使用：

pip install "python-a2a[mcp]"

查看 GitHub 上的完整示例，了解 A2A + MCP 的实际运行情况，或者使用本文中的代码构建你的第一个集成人工智能系统。