构建真正有效的连接型人工智能系统的挑战
如果你正在构建人工智能应用,这种情况可能听起来很熟悉:
你需要特定的人工智能能力来解决业务问题。你找到了完成每个单独任务的出色工具。但把所有东西连接在一起却占据了大部分开发时间,还创建了一个脆弱的系统,一旦有任何更改就会崩溃。
好消息是,有一个今天就能用的实用解决方案。
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 FastMCPAgent
class 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 re
                numbers = [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 requests
    import re
    query = 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 yf
    try:
        # 获取股票数据
        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 re
            company_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 re
            ticker_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 os
class 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)
        # 对于所有其他消息,委托给 OpenAI
        return 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 argparse
def 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 架构今天就能带来实实在在的实用价值:
- 大幅减少集成工作:无需为每个服务编写自定义连接器——一切都能说同一种语言。
- 轻松替换组件:需要更好的数据源?只需插入一个新的智能体——协调器保持不变。
- 清晰的错误边界:每个智能体处理自己的错误,防止系统性故障。
- 简单可扩展:想添加历史数据?只需创建一个新的 MCP 工具,并通过新的或现有的智能体暴露它。
- 跨项目复用组件:那个 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 的实际运行情况,或者使用本文中的代码构建你的第一个集成人工智能系统。
















![[密码学基础]GB与GM国密标准深度解析:定位、差异与协同发展](https://i-blog.csdnimg.cn/direct/c0887140f6664f3bb49b9915ad087560.png#pic_center)


