这不很多人MCP还没拎得清楚，A2A又来了，关键这俩长的还有点像，今天就来仔细说说这个谷歌的新玩意 - Agent2Agent（A2A）协议。

• 为什么会诞生A2A？
• A2A到底是什么？
• A2A架构是怎样的？
• A2A和MCP什么关系？
• A2A的Demo新体验

1、为什么会诞生A2A？

让我们假想一个简单的场景：你开发了某个Agent（比如Web自动化机器人或AI研究员），现在我在开发一个新的Agent系统，希望可以把部分任务交给你的Agent来完成，我们会怎么做呢？

• 最简单的，你可以把Agent代码交给我，我把这个Agent转化成一个函数工具为我的Agent使用，这是可行的（甚至可以用上MCP）：
• 我还可以借助多Agent开发框架，直接构建分布式多Agent系统。这也是可行的，比如利用AutoGen框架：

但是这样的集成一方面需要大量代码级的改造，另一方面耦合度太高了。且不论不同的开发平台和框架可能的兼容性问题，后期更新维护就是大问题。

所以一个更简单的办法就是开放API，于是协商了下接口参数，很快做成了这样：

现在这个方案貌似还不错，可能也是一些有前后台的Agent系统的方案。包括一些Agent构建平台也支持这种模式，比如你可以借助API调用Coze/Dify上创建的Agent。但如果把这样的需求泛化到更多的任务场景，仍然会面临新的挑战：

• 如果扩展到多个Agent，这些Agent有不同的供应商、不同的平台、不同的开发框架，它们之间有协作的需求。那么就会面临和MCP之前一样的问题：你需要去做大量的API集成，不同的消息格式与通信协议，需要你来做“粘合”。
• Agent系统的真实任务场景复杂度往往较高，简单的无状态API无法满足。比如：

* 一个需要很长时间运行的任务

* 多轮对话的问题，Human-in-the-loop工作流

* 流式输出的问题

* 不同供应商之间的Agent之间的互信问题

这种需要让不同类型的AI Agent之间（也可以是其他应用与Agent之间）实现更标准的通信与互操作的需求，就是A2A诞生的原因。

2、A2A到底是什么？

A2A（Agent2Agent）是一种开放协议，旨在实现不透明Agent之间的通信和互操作。其主要作用就是为不同的Agent提供一种共同的“语言”标准，在它们需要“交流”的时候可以更加简单、高效与安全，实现信息共享与任务分配，从而更快的完成任务，即使它们的底层平台完全不同：

整体上而言，A2A提供的价值与MCP是类似的：

• 降低异构Agent之间的集成复杂性：你无需了解对端Agent的细节。
• 提高Agent能力的可复用性：你可以把某个任务交给更擅长它的Agent。
• 更好的扩展性以适应变化：Agent的内部逻辑变化可以被A2A所隔离。

A2A协议对Agent之间集成的如下方面进行标准化：

• 集成架构与关键组件
• 消息与通信机制（JSON-RPC2.0与HTTP）
• 服务端与客户端的功能规范
• 安全验证与授权机制

3、A2A架构是怎样的？

也就是“基于A2A规范的Agent集成架构”，大致描述如下：

几个关键的组件构成如下：

• Agent Card：这是Agent的“名片”，代表“我有怎样的任务完成能力“。
• A2A Server：访问对端Agent的的入口，基于Web Server运行，提供端点供客户端发起各种请求并给予响应；也会通过连接发起主动通知、任务结果给客户端。
• A2A Client：访问A2A Server的其他Agent或定制应用。所以Client与Server是相对的，一个客户端Agent同时也可以是其他Agent的服务端。
• Task（任务）：这是客户端交给服务端Agent需要完成的工作任务。任务过程可能需要客户端的协作；任务结果可以同步等待也可以异步获取。
• Artifact（工件）：服务端Agent的最终产出被称为Artifact。比如一段文字、一个报告、一个图片或者视频；一次任务可以产生多个Artifact。
• Messages（消息）：是指的客户端与服务端Agent之间的多轮沟通内容，存在两个角色：user和agent，内容包括各种上下文、指令、中间状态等。客户端向Agent发送的所有内容都是消息；Agent也可以发送消息给客户端传递状态或指令（比如要求客户端补充信息），但服务端Agent的最终结果则是用Artifact来表示。
• Notifications（通知）：服务端Agent向客户端主动推送的信息，比如一个长期任务的中间运行状态，一个任务有了新的产出等。客户端可以设置通知处理回调。

• 所以基本的运行过程就是：
• 服务端Agent启动Server，通过Agent Card展示能力
• 客户端Agent/应用连接服务端，并查看Agent Card
• 根据Agent Card可以把后续的任务发送给服务端Agent
• 服务端Agent处理任务，必要的时候会输出中间结果，或者要求客户端补充信息
• 客户端可以等待任务完成；也可以根据任务ID异步获取任务结果
• 客户端也可以设置通知回调，及时获得任务的进展与工作成果

4、A2A与MCP什么关系？

A2A在诞生动机、架构甚至协议方面都与MCP非常相似，但它们之间的关系与区别还是很清楚的：

MCP解决的是Agent与外部工具/数据之间的集成，是Agent的“内部事务”；A2A解决的是Agent与Agent之间的集成，属于更高层次的集成关系。

它们之间是可以共存与协作的，比如：

在这样的架构中Agent通过MCP使用工具，Agent与Agent之间则通过A2A产生互动与协作。

我们对两者做个简单的对比：

5、A2A的Demo新体验

A2A还没有完整的官方SDK放出，但是谷歌公司在项目中给出了几个帮助理解A2A的示例，我们对其中一个做了简单修改以演示A2A的能力。

【首先准备一个Agent】

这是一个LangGraph开发的智能体，可以独立运行。你需要实现的主要接口是invoke与stream，分别用于非流式与流式调用（在返回消息格式上有特殊处理）。其“轮廓”长这样，内部逻辑无论你是用ReAct Agent还是定义Workflow都不影响：

...
@tool
def search_tavily(query: str, search_depth: str = "basic") -> Dict[str, Any]:
    ...

class ResponseFormat(BaseModel):
    """以这种格式回应用户。"""
    status: Literal["input_required", "completed", "error"] = "input_required"
    message: str

class SearchAgent:

    SYSTEM_INSTRUCTION = (
        "你是一个专门进行网络搜索的助手。"
        "你的主要目的是使用'search_tavily'工具来回答用户问题，提供最新、最相关的信息。"
        "如果需要用户提供更多信息来进行有效搜索，将响应状态设置为input_required。"
        "如果处理请求时发生错误，将响应状态设置为error。"
        "如果请求已完成并提供了答案，将响应状态设置为completed。"
    )
     
    def __init__(self):
        ...
    def invoke(self, query, sessionId) -> str:
        ...

    async def stream(self, query, sessionId) -> AsyncIterable[Dict[str, Any]]:
        ... 

先写个本地程序对Agent做测试，确保可用：

【启动A2A Server】

现在给这个Agent增加一个前置A2A Server，让它可以对外提供服务，主要逻辑长这样：

def main(host, port):
    
    try:
        capabilities = AgentCapabilities(streaming=True, pushNotifications=True)
        skill = AgentSkill(
            id="search_web",
            name="搜索工具",
            description="搜索web上的相关信息",
            tags=["Web搜索", "互联网搜索"],
            examples=["请搜索最新的黑神话悟空的消息"],
        )

        agent_card = AgentCard(
            name="搜索助手",
            description="搜索Web上的相关信息",
            url=f"http://{host}:{port}/",
            version="1.0.0",
            defaultInputModes=SearchAgent.SUPPORTED_CONTENT_TYPES,
            defaultOutputModes=SearchAgent.SUPPORTED_CONTENT_TYPES,
            capabilities=capabilities,
            skills=[skill],
        )

        notification_sender_auth = PushNotificationSenderAuth()
        notification_sender_auth.generate_jwk()
        server = A2AServer(
            agent_card=agent_card,
            task_manager=AgentTaskManager(agent=SearchAgent(), notification_sender_auth=notification_sender_auth),
            host=host,
            port=port,
        )

        server.app.add_route(
            "/.well-known/jwks.json", notification_sender_auth.handle_jwks_endpoint, methods=["GET"]
        )

        logger.info(f"正在启动服务器，地址：{host}:{port}")
        server.start()
    except MissingAPIKeyError as e:
        logger.error(f"错误：{e}")
        exit(1)
    except Exception as e:
        logger.error(f"服务器启动过程中发生错误：{e}")
        exit(1)


if __name__ == "__main__":
    main()

基本过程是声明Agent的能力->发布成Agent卡片->启动A2AServer（uvicorn)。其核心是一个借助于Agent来完成任务的task_manager。

现在启动这个Server：

【客户端应用测试】

用官方提供的客户端命令行来测试这个Server。可以看到，客户端启动时会读取服务端的Agent卡片，了解到Agent的能力与特性（比如Streaming代表支持流式）：

现在向这个Agent发送一个任务：

客户端采用了Streaming输出，所以这里可以看到一些中间输出。这些输出可以帮助理解A2A协议的一些标准，比如：

• 中间过程都是用message来传递，发送的角色是agent
• 最终的任务输出则是artifact，代表这是任务的最后产出
• 不管是message还是artifact，具体内容都是有多个part组成
• 
  

我们来测试几个能力：

* 多轮连续对话

我们输入“我要搜索”，可以看到收到的消息状态是"input-required"，这是A2A协议规定的任务状态之一，代表需要补充信息。通过这种方式，你后续的消息会被追加进入任务输入（这也需要Agent支持），实现多轮对话：

* 服务端通知消息

在客户端可以启用通知推送，这时候会在客户端启动一个Server用来等待服务端的通知消息，服务端可以会通过post把任务消息推送到客户端：

* 异步任务结果获取

A2A的任务可以完全异步，即使客户端断开也不影响获取任务结果。为了证明这一点，我们另写了一个简单的根据task_id轮询任务结果的程序。然后执行如下步骤：

1. 在Agent中增加一段sleep时间，模拟很长的任务执行时间；然后启动Server；

2. 在官方客户端中启动一个新任务，然后很快强行退出客户端：

3. 拷贝出任务ID（任务ID是客户端生成的，可以直接在客户端打印出来）。

4. 启动我们编写的轮询客户端（带上任务ID），会看到如下结果：

可以看到，尽管发起任务的客户端已经退出，但不会影响你获取任务结果。

这个Demo很好的演示了A2A的一些原理与能力，包括能力发布、流式处理、多轮对话、异步任务等。显然，你也可以把A2A Server作为企业智能体系统的前后端通信的方式（取代FastAPI），可以减少很多复杂问题的工作量。

以上就是今天要介绍的A2A协议的全部。由于A2A还是个刚刚落地的“新生儿”，未来相信还会不断完善与增强，让我们一起期待官方正式发布SDK吧。

更多免费AI功能 云片AI：https://y-p.cc/?f=tt

本文完，记得随手点个赞、收藏和转发三连，大家感兴趣的可以关注下，后续我再研究点新东西分享给大家~

关注后私信发送：DeepSeek，即可领取完整版资料