1. Agent2Agent 协议

2025年4月9日，Google 正式发布了 Agent2Agent Protocol （以下简称 “A2A”）。该协议为不同类型的智能体之间搭建了一座高效沟通与协作的桥梁，无论是独立Agent与独立Agent、独立Agent与企业Agent，亦或是企业Agent与企业Agent，都能借助该协议实现通信交互和协作。

算是比较新的技术，有

1.1 四大特性

（1）Secure collaboration (安全协作): A2A协议全程保障智能体间的通信与数据安全，通过身份验证、数据加密和访问控制三重机制，有效防范未授权访问及数据泄露风险。

（2）Task and state mgmt (任务与状态管理): 这表明协议能够高效管理任务的全生命周期及其状态流转。例如，从任务提交开始，直到最终完成、失败或被取消，协议都能全程跟踪并妥善处理。

（3）UX negotiation (用户体验协商):该功能支持智能体间协作协商，通过动态调整服务策略来持续优化用户体验。

（4）Capability discovery (能力发现): A2A 协议实现了智能体之间的自动能力发现机制。通过 AgentCard 这类"数字名片"，智能体可以主动发布自身功能，而其他智能体则能基于这些公开信息筛选最合适的协作伙伴。

1.2 Agent2Agent 核心概念

Client Agent 和 Server Agent 交互的过程中，会涉及到一些 Entity：AgentCard、Task 、Artifact等，下面做个介绍。

1.2.1 AgentSkill

AgentSkill描述代理的具体能力或功能模块，例如处理特定任务的技能。它包括技能名称、描述、示例、输入/输出模式等。在A2A协议中，技能是代理卡片（AgentCard）的组成部分，用于细粒度服务发现。支持扩展标签和示例，便于代理间匹配调用。

代码示例：

from python_a2a import  AgentSkill
# 定义一个代理技能
ticket_skill = AgentSkill(
    name="book_ticket",
    description="预订火车票的技能",
    examples=["预订从上海到北京的火车票"],
    input_modes=["text/plain"],
    output_modes=["text/plain"]
)

print(ticket_skill)
print(ticket_skill.to_dict())

运行结果：

AgentSkill(name='book_ticket', description='预订火车票的技能', id='321ec43d-043f-4934-9046-56f730b33356', tags=[], examples=['预订从上海到北京的火车票'], input_modes=['text/plain'], output_modes=['text/plain'])
{'id': '321ec43d-043f-4934-9046-56f730b33356', 'name': 'book_ticket', 'description': '预订火车票的技能', 'tags': [], 'examples': ['预订从上海到北京的火车票'], 'inputModes': ['text/plain'], 'outputModes': ['text/plain']}

1.2.2 AgentCard

AgentCard是A2A协议中定义Server Agent元数据的核心组件，作为代理发现和服务注册的关键载体。该卡片详细记录了代理的基础信息（名称、描述、URL、版本）、功能特性（技能列表、流式传输支持等）以及输入输出模式等关键参数。

在A2A生态系统中，AgentCard承担着服务入口点的角色：一方面支持JSON序列化以实现高效网络传输；另一方面为Client Agent提供决策依据，使其能够基于不同Server Agent的能力描述（包括认证机制等）智能选择最适合的任务执行对象。

代码示例:

from python_a2a import AgentCard, AgentSkill
# 创建一个代理技能
ticket_skill = AgentSkill(
    name="book_ticket",
    description="预订火车票的技能",
    examples=["预订从上海到北京的火车票"],
    input_modes=["text/plain"],
    output_modes=["text/plain"]
)
# 创建代理卡片
agent_card = AgentCard(
    name="TicketAgent",
    description="一个可以预订票务的代理",
    url="http://127.0.0.1:5009",
    version="1.0.0",
    skills=[ticket_skill],
    capabilities={"streaming": True}
)
# 打印代理卡片的字典表示（用于序列化）
print(agent_card)
print(agent_card.to_dict())

输出示例：

AgentCard(name='TicketAgent', description='一个可以预订票务的代理', url='http://127.0.0.1:5009', version='1.0.0', authentication=None, capabilities={'streaming': True}, default_input_modes=['text/plain'], default_output_modes=['text/plain'], skills=[AgentSkill(name='book_ticket', description='预订火车票的技能', id='487e3ee8-1cbd-4567-b083-a9627a486d1c', tags=[], examples=['预订从上海到北京的火车票'], input_modes=['text/plain'], output_modes=['text/plain'])], provider=None, documentation_url=None)
{'name': 'TicketAgent', 'description': '一个可以预订票务的代理', 'url': 'http://127.0.0.1:5009', 'version': '1.0.0', 'capabilities': {'streaming': True}, 'defaultInputModes': ['text/plain'], 'defaultOutputModes': ['text/plain'], 'skills': [{'id': '487e3ee8-1cbd-4567-b083-a9627a486d1c', 'name': 'book_ticket', 'description': '预订火车票的技能', 'tags': [], 'examples': ['预订从上海到北京的火车票'], 'inputModes': ['text/plain'], 'outputModes': ['text/plain']}]}

1.2.3 Task

Task是由Client Agent创建并管理的实体，具有明确的状态属性，其状态更新由Server Agent负责维护。每个Task都对应特定的目标或结果。在执行过程中，Client Agent和Server Agent通过Message进行通信，Server Agent执行后产生的输出结果称为Artifact。

每个Task都包含唯一的sessionId标识。多个Task可共享相同sessionId，表示这些Task属于同一个Session，便于统一管理和跟踪相关任务的执行流程。

代码示例:

from python_a2a import Task, Message, MessageRole, TextContent

# 创建任务
message = Message(content=TextContent(text="查询天气"), role=MessageRole.USER)
task = Task(message=message.to_dict())
print(task)

运行结果：

Task(id='5e63778f-a912-4f07-a757-4b47614d0d1b', session_id='7211f559-d616-4f56-82d2-c74976f86390', status=TaskStatus(state=<TaskState.SUBMITTED: 'submitted'>, message=None, timestamp='2025-10-24T23:20:44.444053'), message={'content': {'text': '查询天气', 'type': <ContentType.TEXT: 'text'>}, 'role': 'user', 'message_id': '8a3c1015-0871-4886-b70d-cf1434ad1caf'}, history=[], artifacts=[], metadata={})

1.2.4 TaskState

TaskState 枚举类

TaskState 是一个枚举类，用于定义任务的各种可能状态，包括：

• SUBMITTED（已提交）
• COMPLETED（已完成）
• FAILED（已失败）

作为任务生命周期管理的基础组件，该枚举类确保了状态的一致性和代码的可读性。

TaskState 状态表格：

状态名称	值	中文描述
SUBMITTED	submitted	任务已提交，等待处理。
WAITING	waiting	任务正在等待，例如等待外部资源或输入。
INPUT_REQUIRED	input-required	任务需要额外用户输入以继续执行。
COMPLETED	completed	任务已成功完成，结果可用。
CANCELED	canceled	任务被取消，未完成执行。
FAILED	failed	任务执行失败，可能包含错误信息。
UNKNOWN	unknown	未知状态，通常用于处理无效或未识别的状态。

注意：

状态名称：TaskState 枚举成员（如 TaskState.SUBMITTED），提供类型安全的代码引用
值：对应的字符串值（如"submitted"），用于序列化或系统间交互
中文描述：说明各状态在任务生命周期中的功能和使用场景，便于开发者理解

代码示例：

from python_a2a import TaskState  # 只需相关导入
# 检查任务状态
if TaskState.COMPLETED == "completed":
    print("任务完成")
state = TaskState.SUBMITTED
print("转换后的状态值：", state.value)
print(state)

运行结果：

任务完成
转换后的状态值： submitted
TaskState.SUBMITTED

1.2.5 TaskStatus

TaskStatus 是用于表示 A2A 任务当前状态的对象，包含三个关键属性：

1. 状态枚举（TaskState）
2. 附加消息
3. 时间戳

该对象主要用于跟踪任务执行进度，支持序列化和格式转换功能，能够动态反映任务处理状态。

使用注意事项：

• TaskStatus 必须包含一个有效的 TaskState 作为其 state 字段
• TaskState 是 TaskStatus 的必要依赖项

示例代码展示了 TaskState 和 TaskStatus 的典型结合使用方法。

代码示例：

from python_a2a import TaskStatus, TaskState

status_completed = TaskStatus(
    state=TaskState.COMPLETED,
    message={"info": "任务成功完成"}
)

status_failed = TaskStatus(
    state=TaskState.FAILED,
    message={"error": "无法处理请求"}
)

# 打印字典表示
print("完成状态：", status_completed.to_dict())
print("失败状态：", status_failed.to_dict())

输出日志：

完成状态： {'state': 'completed', 'timestamp': '2025-10-24T22:55:37.883274', 'message': {'info': '任务成功完成'}}
失败状态： {'state': 'failed', 'timestamp': '2025-10-24T22:55:37.883274', 'message': {'error': '无法处理请求'}}

1.2.6 A2AServer

A2AServer作为A2A协议的核心实现类，负责构建代理服务器。它继承自BaseA2AServer，主要功能包括任务处理（handle_task）、消息处理（handle_message）和路由配置（setup_routes）。该类集成了任务存储管理、流式订阅支持，并兼容Google A2A模式，通过Flask路由支持、任务处理逻辑和错误处理机制，确保代理间通信的可靠性。

任务处理说明：

当继承A2AServer时，系统会自动处理任务请求。通常情况下无需手动创建Task对象，A2AServer会自动将传入请求解析为Task对象，并传递给handle_task方法进行处理。

handle_task方法功能： 该方法负责解析任务输入、执行查询处理、封装处理结果，并返回最终的task任务对象。

代码示例：

from python_a2a import A2AServer, run_server, AgentCard, AgentSkill, TaskStatus, TaskState

# 定义代理卡片
ticket_card = AgentCard(
    name="TicketAgentServer",
    description="票务代理",
    url="http://127.0.0.1:5009/a2a",
    skills=[AgentSkill(name="book_ticket", description="预订票务")]
)

# 自定义 A2AServer 子类
class TicketServer(A2AServer):
    def __init__(self):
        super().__init__(agent_card=ticket_card)

    def handle_task(self, task):
        print(f"任务状态：{task.status.state}")
        return task

# 启动服务器
if __name__ == "__main__":
    server = TicketServer()
    print(f"[{server.agent_card.name}] 创建服务成功")
    run_server(server, host="127.0.0.1", port=5009, debug=False)

1.2.7 artifacts

在 A2A 协议中，artifacts 是 Task 对象的核心字段，用于存储任务执行后生成的输出结果。该字段是一个列表结构，其中每个元素都是一个产物对象，通常以字典形式存储，并包含指向多个内容片段的 "parts" 键。

artifacts 作为任务结果的结构化容器，支持存储文本内容、函数调用结果和错误信息等多种数据类型，确保客户端能够准确解析并有效利用代理生成的输出内容。

task.artifacts = [
    {
        "parts": [
            {"type": "text", "text": "处理结果"},
            {"type": "error", "message": "错误描述"},
            {"type": "function_response", "name": "func_name", "response": {...}}
        ]
    }
]

常见类型：

• text：纯文本结果，如用户查询的响应。
• error：错误信息，包含错误描述。
• function_response：函数调用结果，包含函数名和返回数据。"name" 和 "response" 用于函数响应。
• function_call：发起的函数调用，包含函数名和参数。

代码示例

服务端：

from python_a2a import A2AServer, run_server, AgentCard, AgentSkill, TaskStatus, TaskState

# 定义代理卡片
ticket_card = AgentCard(
    name="TicketAgentServer",
    description="票务代理",
    url="http://127.0.0.1:5010",
    skills=[AgentSkill(name="book_ticket", description="预订票务")]
)

# 自定义 A2AServer 子类
class TicketServer(A2AServer):
    def __init__(self):
        super().__init__(agent_card=ticket_card)

    def handle_task(self, task):
        print("收到A2A任务的task:=>", task)
        #默认写法：获取任务内容
        query = (task.message or {}).get("content", {}).get("text", "")

        if "上海" in query and "北京" in query:
        # 这里的结果可以来自于 MCP 模块，这里我们直接模拟结果
            train_result = "上海到北京的火车票已经预订成功！  G1001,10车1A "
        else:
            train_result = "请输入明确的出发地和目的地。"

        task.artifacts = [{"parts": [{"type": "text", "text": train_result}]}]
        task.status = TaskStatus(state=TaskState.COMPLETED)
        print(f"[{self.agent_card.name} 日志] 任务处理完毕")
        print(f"[{self.agent_card.name} 日志] 输出结果task: {task}")
        print(f"[{self.agent_card.name} 日志] 输出结果task.artifacts: {task.artifacts}")
        return task

# 启动服务器
if __name__ == "__main__":
    server = TicketServer()
    print(f"[{server.agent_card.name}] 启动成功，服务地址: {server.agent_card.url}")
    run_server(server, host="127.0.0.1", port=5010, debug=True)

客户端：

import asyncio
from python_a2a import A2AClient

async def main():
    ticket_client = A2AClient("http://127.0.0.1:5010")

    #预订火车票
    ticket_query = "预订一张从北京到上海的火车票"
    print(f"[主控客户端日志]预订票务 -> '{ticket_query}'")
    ticket_result = ticket_client.ask(ticket_query)
    print(f"[主控客户端日志] 收到票务预订结果: {ticket_result}")

if __name__ == "__main__":
    asyncio.run(main())

先启动服务端，然后启动客户端，最终得到的结果如下：

[主控客户端日志]预订票务 -> '预订一张从北京到上海的火车票'
[主控客户端日志] 收到票务预订结果: 上海到北京的火车票已经预订成功！  G1001,10车1A 

1.2.8 AgentNetwork

AgentNetwork 是 A2A 协议中的核心网络管理类，主要负责集中管理和发现 A2A 兼容代理。其主要功能包括：

1. 维护代理列表
2. 支持通过 URL 或客户端实例添加代理
3. 缓存代理元数据（如 AgentCard）

核心价值：

• 简化多代理协作流程
• 提供完整的代理管理功能（添加/获取/列表/发现/移除）
• 适用于构建分布式代理系统
• 消除手动管理多个客户端的复杂性

核心特性：

• 添加代理：通过 add 方法，支持 URL（自动创建 A2AClient）或现有客户端。
• 代理元数据：自动缓存 AgentCard，便于查询代理能力。
• 发现代理：通过 discover_agents 从 URL 列表自动添加有效代理。
• 扩展性：支持头信息（headers）和异常处理。

代码示例：

from python_a2a import AgentNetwork
network = AgentNetwork(name="MyNetwork")
network.add("TicketAgent", "http://127.0.0.1:5010")

print(f"agent network-->{network.agent_cards}")
print('*'*80)

# 调用
client = network.get_agent("TicketAgent")
print(client.ask("预订一张从北京到上海的火车票"))

运行结果：

INFO:python_a2a.client.network:Added agent 'TicketAgent' from URL: http://127.0.0.1:5010
agent network-->{'TicketAgent': AgentCard(name='TicketAgentServer', description='票务代理', url='http://127.0.0.1:5010', version='1.0.0', authentication=None, capabilities={'google_a2a_compatible': True, 'parts_array_format': True, 'pushNotifications': False, 'stateTransitionHistory': False, 'streaming': True}, default_input_modes=['text/plain'], default_output_modes=['text/plain'], skills=[AgentSkill(name='book_ticket', description='预订票务', id='65d84e15-c135-4ffd-9a60-16caf46d88e9', tags=[], examples=[], input_modes=['text/plain'], output_modes=['text/plain'])], provider=None, documentation_url=None)}
********************************************************************************
火车票已经预订成功！  G1001,10车1A 

1.2.9 AIAgentRouter

定义：

AIAgentRouter 是一个定义在 router.py 文件中的智能路由类，它通过分析查询意图和上下文信息，自动选择最合适的代理进行处理。

作用：

多代理网络支持智能路由功能，可根据语义分析、历史上下文及缓存机制，自动将用户查询精准匹配至最适合的代理节点。系统以代理名称及置信度作为响应格式，实现代理选择的自动化处理。

核心特性：

• LLM 驱动：使用 LLM 客户端（如 ChatOpenAI）生成路由提示，分析查询匹配代理描述和技能。
• 上下文支持：包含对话历史（max_history_tokens 限制令牌数）。
• 缓存优化：类似查询使用缓存减少 LLM 调用。
• 回退机制：LLM 失败时，使用关键词匹配回退路由。
• 系统提示：自定义提示指导路由决策。

代码示例：

from python_a2a import AIAgentRouter, AgentNetwork
from langchain_openai import ChatOpenAI
from agent_learn.config import Config

conf=Config()

# 创建网络
network = AgentNetwork(name="MyNetwork")
network.add("TicketAgent", "http://127.0.0.1:5010")

# 创建模型
llm = ChatOpenAI(base_url=conf.base_url,
                 api_key=conf.api_key,
                 model=conf.model_name,
                 temperature=0.1)

# 创建路由器
router = AIAgentRouter(llm_client=llm, agent_network=network)
agent_name, confidence = router.route_query("预订票")
print(agent_name, confidence)

输出结果：

INFO:python_a2a.client.network:Added agent 'TicketAgent' from URL: http://127.0.0.1:5010
TicketAgent 0.1

2. MCP协议

MCP（Model Context Protocol，模型上下文协议）是Anthropic于2024年1月推出的开放协议，旨在为大型语言模型（LLM）与外部数据源及工具之间建立安全、高效的双向连接通道。

Anthropic致力于将MCP协议打造成为AI领域的"通用接口"，类似于互联网中的HTTP协议。通过MCP协议，各类工具和数据源能够实现无缝对接，最终达成广泛互联的生态系统。

2.1 MCP 核心概念

MCP协议有两个核心角色：客户端与服务端。

MCP服务端 (Tool Provider)

• 角色：工具的提供者。
• 职责：将一个或多个本地函数（例如，Python函数）包装起来，通过一个标准的MCP接口暴露出去。它监听来自客户端的请求，执行对应的函数，并返回结果。
• 例子：一个天气查询服务、一个计算服务、一个数据库访问服务。

MCP客户端 (Tool Consumer)

• 角色：工具的调用者或消费者。
• 职责：连接到MCP服务端，查询可用的工具列表（自发现），并根据需要调用这些工具。
• 例子：大模型Agent、自动化脚本、任何需要远程执行功能的应用程序。

3. 区别与联系

简单来说：MCP 是“给 AI 装工具”，而 A2A 是“让 AI 找外援”。

特性	MCP (Model Context Protocol)	A2A (Agent-to-Agent)
角色定义	Tool (工具)：被动等待调用。	Agent (代理)：具备独立思考和任务管理能力。
Token 消耗	所有的工具描述和中间过程都在主 AI 的上下文中。	主 AI 只负责中转任务，子 Agent 独立消耗 Token。
复杂任务处理	难以处理需要长时间异步运行或多步确认的任务。	天生支持异步任务和复杂的内部流程。
解耦程度	高度依赖主模型的指令遵循能力。	完全解耦。子 Agent 可以用不同的模型、不同的语言实现。
权限隔离	主 AI 拥有所有工具的访问权，存在安全风险。	权限被封装在子 Agent 内部，主 AI 只能看到最终结果。

3.1 为什么不只用多个 MCP？

想象一下你要开发一个“公司全能助理”：

• 全 MCP 方案： 你给一个 LLM 挂载了 50 个 MCP 工具（财务、考勤、订票、代码、数据库）。

  • 痛点： LLM 的上下文会被海量的工具描述塞满（Context Window 爆炸），且工具越多，LLM 越容易产生“幻觉”，分不清该用哪一个。

• A2A 方案： 主助理只挂载 5 个“垂直领域的 Agent”。

  • 优势： 当你说“帮我报销并订张票”时，主助理只需分发任务。

  • 具体到你的代码： TicketServer 不仅仅是一个函数，它是一个独立的服务节点。它可以有自己的排队系统、任务状态（TaskStatus）监控，甚至在订票失败时自动尝试另一种方案，而不需要主 AI 一直盯着它看。

综上所述：还是取决于具体的任务

• 使用 MCP： 当你有一些简单的、原子化的功能（如：查天气、读文件、算数学、查数据库一行数据）。

• 使用 A2A： 当你需要构建一个复杂的分布式系统，其中每个模块都需要独立维护状态、拥有独立逻辑，或者你希望通过“专家分工”来降低主模型的推理负担。