A2A 与 MCP

我们用一个通俗易懂的例子来解释A2A协议和MCP，以及它们之间的关系和如何使用。

想象一下：一个智能的团队和他们的工具

假设你是一个公司的老板，你的公司有很多聪明的"员工"（AI Agent），他们各有所长，比如：

• 小秘书Agent：负责安排会议、管理日程。
• 销售Agent：负责跟客户沟通、推荐产品。
• 数据分析Agent：负责分析销售数据、生成报告。

这些Agent非常智能，但一开始它们之间是独立的，互相不知道对方能做什么，也不知道怎么协作。同时，它们也需要使用一些"工具"才能完成工作，比如：

• 日历应用：用来安排会议。
• CRM系统：用来记录客户信息。
• 数据库：用来存储销售数据。
• PPT制作软件：用来生成报告。

A2A协议：Agent 之间的"通用语言"和"协作规则"

A2A (Agent-to-Agent) 协议就像是给这些智能员工制定了一套**"通用语言"和"协作规则"**。

在没有A2A协议之前，如果小秘书Agent想让销售Agent去联系客户，它们可能需要提前约定好"暗号"或者专门为这次合作开发一套接口。这就像不同的国家有不同的语言，要沟通就得找翻译，或者每次都为特定交流学习新的语言。

有了A2A协议，就像给所有Agent培训了一套**"国际通用语"（比如英语）和一套"团队协作规范"**。

• Agent Card（代理卡）：每个Agent都会有一个"名片"或者"职责说明书"，上面清楚地写着它的名字、能做什么事情（比如"我能安排会议"、"我能分析数据"）、以及如何联系它。这个"名片"就是Agent Card，它让其他Agent可以快速发现并理解它的能力。
• 标准化沟通：A2A规定了Agent之间如何发送请求、如何回复、如何传递任务进度、如何分享文件等等。这就像团队成员之间使用统一的会议流程、报告模板和文件共享方式，让协作变得顺畅高效。

简单来说，A2A协议就是让不同的AI Agent能够互相发现、理解和协同工作的一套标准。

如何使用A2A协议？

如果你要开发Agent，并希望它们能互相协作，你可以：

1. 定义Agent Card：为你的Agent创建一个Agent Card，清楚说明它的能力和接口。
2. 遵循A2A规范：让你的Agent在与其他Agent沟通时，按照A2A协议规定的格式和流程来发送和接收信息。
3. 使用SDK（软件开发工具包）：许多平台会提供A2A的SDK，让你更方便地实现Agent之间的通信。

MCP：Agent 访问"外部工具"的"统一接口"

MCP (Model Context Protocol) 就像是给每个智能员工配备了一个**"万能插头"**，让他们可以方便地连接和使用各种"外部工具"（比如日历、CRM、数据库）。

在没有MCP之前，小秘书Agent可能需要针对日历应用写一套专门的代码来连接它，销售Agent又要为CRM系统写一套完全不同的代码。这就像不同的电器需要不同的插头，非常不方便。

有了MCP，就像给所有Agent都配了一个"USB-C接口"：

• 统一接口：无论外部工具是日历应用、CRM系统还是数据库，Agent都可以通过MCP提供的统一方式（比如JSON-RPC）来调用它们。
• 上下文提供：MCP还可以帮助Agent获取和理解使用工具所需的"上下文信息"，比如"当前是几点"、"要查询的是哪个客户的数据"等等。

简单来说，MCP就是让AI Agent能够以标准化的方式，更方便、更智能地使用各种外部工具和数据。

如何使用MCP？

如果你要让Agent使用外部工具，你可以：

1. 创建MCP Server：为你的外部工具创建一个MCP Server，它会把工具的功能包装成MCP可以理解的接口。
2. 定义工具能力：在MCP Server中定义这个工具能做什么、需要什么参数、返回什么结果。
3. Agent调用MCP Server：你的Agent通过MCP协议调用这个MCP Server，从而间接地使用外部工具。

A2A协议和MCP的关系：紧密合作，共同完成任务

A2A和MCP并不是互相竞争，而是互补的。它们就像一个优秀的团队：

• A2A负责"团队内部协作"：Agent之间互相交流、分配任务、共享进展。
• MCP负责"与外界交互"：每个Agent通过MCP去使用各种外部工具，获取数据，或者执行操作。

想象一个场景：

老板（用户）对小秘书Agent说："帮我安排一个下周二下午两点，关于新产品发布会的会议，邀请销售Agent和数据分析Agent参加。"

1. **小秘书Agent（A2A客户端）**收到任务。
2. 它决定需要安排会议，所以它会调用自己的MCP来连接日历应用（一个外部工具）。
3. 小秘书Agent会通过A2A协议，向销售Agent和数据分析Agent发送会议邀请（这是Agent之间的沟通）。
4. 销售Agent和数据分析Agent收到邀请后，会通过自己的MCP来检查自己的日历，并回复小秘书Agent（也是通过A2A协议）。
5. 在会议中，老板可能问销售Agent："最近新产品的销售情况如何？"
6. 销售Agent会通过自己的MCP去连接CRM系统，获取客户数据。
7. 销售Agent可能觉得数据不够详细，它会通过A2A协议把数据发送给数据分析Agent，让它生成一个详细的销售报告。
8. 数据分析Agent收到数据后，会通过自己的MCP调用数据库获取更多信息，并使用PPT制作软件生成报告。
9. 最后，数据分析Agent通过A2A协议把报告发回给销售Agent，销售Agent再汇报给老板。

总结来说：

• A2A协议让AI Agent之间能够互相"说话"和"协作"，实现更复杂的任务。
• MCP让AI Agent能够方便地"使用"各种"工具"，拓展它们的能力边界。
• 两者结合，构建了一个强大的AI生态系统，让不同的Agent可以像一个高效率的团队一样，协同使用各种工具来解决问题。

好的，我们来构建一个简单的Python小应用，模拟一个AI代理（Agent）如何使用A2A协议与其他代理协作，以及如何通过MCP使用外部工具。

---

场景设定：智能会议助手

我们来创建一个"智能会议助手"小应用。在这个应用中，我们会有两个Agent：

1. 会议调度Agent (MeetingSchedulerAgent)：负责安排会议，并会用到一个"日历工具"。
2. 参会人通知Agent (ParticipantNotifierAgent)：负责通知参会人员会议信息。

A2A协议的体现

• Agent之间的发现与沟通：会议调度Agent需要知道如何通知参会人，所以它会"发现"并调用参会人通知Agent的能力。
• 标准化消息传递：Agent之间传递会议信息和通知请求，会遵循一套简单的消息格式。

MCP的体现

• 工具封装：我们会模拟一个"日历工具"，会议调度Agent通过MCP的方式来"调用"这个日历工具的功能（比如"添加日程"）。
• 统一接口：无论未来有多少个外部工具，Agent都能通过类似的MCP接口去调用它们。

---

小应用代码示例

我们将使用Python类来模拟Agent和MCP的概念。

import json
import uuid

# --- 模拟 A2A 协议相关 ---

class AgentCard:
    """
    模拟 A2A Agent Card，描述 Agent 的能力。
    """
    def __init__(self, agent_id, name, capabilities):
        self.agent_id = agent_id
        self.name = name
        self.capabilities = capabilities # 列表，描述 Agent 能做什么

    def to_json(self):
        return json.dumps({
            "agent_id": self.agent_id,
            "name": self.name,
            "capabilities": self.capabilities
        }, indent=2)

class A2AMessage:
    """
    模拟 A2A 消息结构，用于 Agent 之间的通信。
    """
    def __init__(self, sender_id, receiver_id, message_type, payload):
        self.message_id = str(uuid.uuid4())
        self.sender_id = sender_id
        self.receiver_id = receiver_id
        self.message_type = message_type # 例如: 'request_schedule_meeting', 'notify_participants'
        self.payload = payload # 消息内容

    def to_json(self):
        return json.dumps({
            "message_id": self.message_id,
            "sender_id": self.sender_id,
            "receiver_id": self.receiver_id,
            "message_type": self.message_type,
            "payload": self.payload
        }, indent=2)

# --- 模拟 MCP 协议相关 ---

class ExternalTool:
    """
    模拟一个外部工具，比如一个简化的日历应用。
    """
    def __init__(self, name):
        self.name = name
        self.calendar_events = []

    def add_event(self, event_details):
        print(f"  [{self.name}工具]: 收到请求，正在添加日程: {event_details['title']}")
        self.calendar_events.append(event_details)
        print(f"  [{self.name}工具]: 日程 '{event_details['title']}' 已添加。")
        return {"status": "success", "event_id": str(uuid.uuid4())}

    def get_events(self):
        return self.calendar_events

class MCPServer:
    """
    模拟 MCP Server，负责将外部工具的能力暴露给 Agent。
    """
    def __init__(self, tool):
        self.tool = tool
        self.available_functions = {
            "add_calendar_event": self.tool.add_event,
            "get_calendar_events": self.tool.get_events
        }

    def call_function(self, function_name, *args, **kwargs):
        if function_name in self.available_functions:
            print(f"  [MCP Server]: 代理正在通过 MCP 调用工具功能: {function_name}")
            return self.available_functions[function_name](*args, **kwargs)
        else:
            print(f"  [MCP Server]: 错误：未知的工具功能 '{function_name}'")
            return {"status": "error", "message": "Unknown function"}

# --- 模拟 Agent ---

class BaseAgent:
    """
    Agent 的基类，包含 Agent ID 和 Agent Card。
    """
    def __init__(self, name, capabilities):
        self.agent_id = str(uuid.uuid4())
        self.name = name
        self.card = AgentCard(self.agent_id, self.name, capabilities)
        self.known_agents = {} # 用于存储已知 Agent 的信息 (A2A发现)
        self.mcp_servers = {} # 用于存储 MCP Server 实例 (MCP连接)

    def register_mcp_server(self, tool_name, mcp_server):
        """注册 MCP Server，连接到外部工具"""
        self.mcp_servers[tool_name] = mcp_server
        print(f"Agent '{self.name}' 已连接到工具 '{tool_name}' 的 MCP Server。")

    def discover_agent(self, agent_card):
        """模拟 Agent 发现另一个 Agent"""
        self.known_agents[agent_card.agent_id] = agent_card
        print(f"Agent '{self.name}' 发现了 Agent '{agent_card.name}'。")

    def send_a2a_message(self, receiver_agent_id, message_type, payload):
        """发送 A2A 消息"""
        if receiver_agent_id not in self.known_agents:
            print(f"错误: Agent '{self.name}' 未知 Agent ID: {receiver_agent_id}")
            return None
        message = A2AMessage(self.agent_id, receiver_agent_id, message_type, payload)
        print(f"\n--- Agent '{self.name}' 发送 A2A 消息给 '{self.known_agents[receiver_agent_id].name}' ---")
        print(message.to_json())
        return message

    def receive_a2a_message(self, message):
        """接收 A2A 消息 (这是一个需要子类实现的抽象方法)"""
        print(f"\n--- Agent '{self.name}' 收到 A2A 消息 ---")
        print(message.to_json())
        return f"Agent '{self.name}' 处理了消息: {message.message_type}"

class MeetingSchedulerAgent(BaseAgent):
    """
    会议调度 Agent，负责安排会议并调用日历工具 (通过 MCP)。
    """
    def __init__(self):
        super().__init__("会议调度Agent", ["schedule_meeting", "use_calendar_tool"])

    def schedule_meeting(self, meeting_details, participants_agent_id):
        print(f"\n### 会议调度Agent: 正在调度会议 '{meeting_details['title']}' ###")

        # 1. 通过 MCP 调用日历工具添加日程
        if "calendar_tool" in self.mcp_servers:
            print(f"会议调度Agent: 通过 MCP 调用 '日历工具'...")
            mcp_server = self.mcp_servers["calendar_tool"]
            response = mcp_server.call_function("add_calendar_event", meeting_details)
            if response["status"] == "success":
                print(f"会议调度Agent: 日程已成功添加。事件ID: {response['event_id']}")
                meeting_details["event_id"] = response["event_id"]
            else:
                print(f"会议调度Agent: 日程添加失败: {response['message']}")
                return False
        else:
            print("会议调度Agent: 未连接到日历工具的 MCP Server。")
            return False

        # 2. 通过 A2A 协议通知参会人通知Agent
        if participants_agent_id in self.known_agents:
            print(f"会议调度Agent: 通过 A2A 通知参会人通知Agent...")
            notification_payload = {
                "meeting_title": meeting_details["title"],
                "time": meeting_details["time"],
                "participants": meeting_details["participants"]
            }
            message = self.send_a2a_message(
                participants_agent_id,
                "notify_meeting_participants",
                notification_payload
            )
            # 在实际应用中，这里会有一个消息队列或Agent服务来处理接收
            # 这里我们直接模拟接收方处理
            if message:
                print(f"会议调度Agent: 收到来自 {self.known_agents[participants_agent_id].name} 的回复: {self.known_agents[participants_agent_id].receive_a2a_message(message)}")
            return True
        else:
            print("会议调度Agent: 无法找到参会人通知Agent。")
            return False

    def receive_a2a_message(self, message):
        super().receive_a2a_message(message)
        if message.message_type == "meeting_details_request":
            print(f"会议调度Agent: 收到会议详情请求，但目前不支持提供。")
            return {"status": "unsupported", "message": "Cannot provide meeting details on request."}
        else:
            return {"status": "success", "message": "Message received and acknowledged."}


class ParticipantNotifierAgent(BaseAgent):
    """
    参会人通知 Agent，负责接收会议信息并"通知"参会人。
    """
    def __init__(self):
        super().__init__("参会人通知Agent", ["notify_participants"])

    def receive_a2a_message(self, message):
        super().receive_a2a_message(message)
        if message.message_type == "notify_meeting_participants":
            meeting_title = message.payload["meeting_title"]
            time = message.payload["time"]
            participants = message.payload["participants"]
            print(f"参会人通知Agent: 收到会议通知请求：")
            print(f"  会议: '{meeting_title}'")
            print(f"  时间: {time}")
            print(f"  参会人: {', '.join(participants)}")
            print(f"参会人通知Agent: 正在模拟发送通知给参会人...")
            return {"status": "success", "message": "Participants notified."}
        else:
            print(f"参会人通知Agent: 收到未知消息类型: {message.message_type}")
            return {"status": "error", "message": "Unknown message type."}

# --- 运行小应用 ---

def run_application():
    print("--- 初始化 AI Agent 和工具 ---")

    # 1. 实例化外部工具 (日历)
    calendar_tool = ExternalTool("我的日历")
    # 2. 实例化 MCP Server，连接日历工具
    calendar_mcp_server = MCPServer(calendar_tool)

    # 3. 实例化 Agent
    meeting_scheduler = MeetingSchedulerAgent()
    participant_notifier = ParticipantNotifierAgent()

    # 4. Agent 连接到 MCP Server (让 MeetingSchedulerAgent 能用日历)
    meeting_scheduler.register_mcp_server("calendar_tool", calendar_mcp_server)

    # 5. Agent 之间互相发现 (模拟 A2A 协议的发现机制)
    meeting_scheduler.discover_agent(participant_notifier.card)
    participant_notifier.discover_agent(meeting_scheduler.card) # 尽管在这个例子中 notifier 不需要调用 scheduler，但发现是双向的

    print("\n--- 开始执行任务 ---")

    # 会议调度 Agent 接收到安排会议的请求
    meeting_details = {
        "title": "季度产品评审会",
        "time": "2025年6月20日 下午3:00",
        "participants": ["Alice", "Bob", "Charlie", "David"]
    }

    # 会议调度 Agent 开始调度会议，并通知参会人
    success = meeting_scheduler.schedule_meeting(
        meeting_details,
        participant_notifier.agent_id
    )

    if success:
        print("\n--- 任务完成 ---")
        print("会议已安排并通知参会人。")
        print("\n--- 查看日历工具中的日程 ---")
        print(f"日历中的事件: {calendar_tool.get_events()}")
    else:
        print("\n--- 任务失败 ---")

if __name__ == "__main__":
    run_application()

---

代码讲解与 A2A / MCP 的体现

1. A2A 协议的体现

• AgentCard 类：
  • 这就是我们说的"名片"或"职责说明书"。每个Agent都有一个，描述了它的agent_id、name和capabilities（能做什么）。
  • 在 run_application 函数中，我们通过 meeting_scheduler.discover_agent(participant_notifier.card) 模拟了Agent之间的互相"发现"和了解对方能力的过程。
• A2AMessage 类：
  • 定义了Agent之间通信的标准格式，包含sender_id（发送者）、receiver_id（接收者）、message_type（消息类型，比如"请求调度会议"、"通知参会人"）和payload（消息内容）。
  • MeetingSchedulerAgent 通过调用 self.send_a2a_message() 来向 ParticipantNotifierAgent 发送通知。
  • ParticipantNotifierAgent 通过 receive_a2a_message() 方法来接收并处理来自其他Agent的消息。
• Agent 之间的协作：
  • MeetingSchedulerAgent 不直接通知参会人，而是将这个任务通过 A2A 消息 发送给 ParticipantNotifierAgent，体现了Agent之间的分工协作。

2. MCP 协议的体现

• ExternalTool 类：
  • 模拟了一个真实的外部工具，这里是一个简化的日历。它有自己的内部逻辑（add_event, get_events）。
• MCPServer 类：
  • 这就是MCP的核心。它作为一个"中间层"，将ExternalTool的具体功能（add_event、get_events）包装成Agent可以通过统一接口调用的"服务"（available_functions字典）。
  • Agent 不直接与 ExternalTool 的具体方法交互，而是通过 MCPServer.call_function() 方法来调用。这提供了统一的工具访问接口。
• Agent 与 MCP 的连接：
  • MeetingSchedulerAgent 通过调用 self.register_mcp_server("calendar_tool", calendar_mcp_server)，将其"连接"到日历工具的MCP Server。这意味着这个Agent现在知道如何通过MCP来使用日历工具。
  • 在 MeetingSchedulerAgent.schedule_meeting 方法中，通过 mcp_server.call_function("add_calendar_event", meeting_details) 实际完成了对日历工具的调用。

---

如何运行这个小应用

1. 将上面的Python代码保存为一个 .py 文件，例如 ai_collaboration_app.py。
2. 打开你的终端或命令提示符。
3. 导航到你保存文件的目录。
4. 运行命令：python ai_collaboration_app.py

运行结果解释

你会看到程序输出一系列日志，清晰地展示了：

1. Agent和工具的初始化：哪些Agent被创建了，哪些MCP Server被连接了。
2. Agent的发现过程：各个Agent"知道"了对方的存在和能力。
3. 任务执行过程：
   • 会议调度Agent 首先通过 MCP 调用了"日历工具"的add_event功能来安排会议。
   • 然后，会议调度Agent 通过 A2A 协议 发送了一条消息给 参会人通知Agent。
   • 参会人通知Agent 接收并处理了这条A2A消息，模拟了通知参会人。
4. 最终结果：确认会议已安排并通知。

---

通过这个小应用，你应该能更直观地理解A2A协议如何促进AI Agent之间的协作与通信，以及MCP如何为AI Agent提供标准化、统一的外部工具访问能力。它们共同为构建更智能、更复杂的AI系统奠定了基础。