1.什么是A2A ?

Google 发布了一个名为 Agent To Agent 的协议,A2A 是一个与 Agent 有关的协议,对于各路人马创建的 Agent,A2A 提供了一种统一的封装方式。这样一来,不同来源的 Agent 能够实现互相调用,从而打破彼此之间的隔阂,避免 Agent 成为孤立的“信息孤岛”,这对推动 Agent 之间的协同合作与生态发展很有价值。你可能会疑惑,已经有了MCP,再出一个A2A协议有必要吗 ? 它们有什么区别、联系 ?用一张图,帮你解惑。
在这里插入图片描述

2. 关键源码解读

Step1:
打开 A2A 的官方 GitHub 地址:https://github.com/a2aproject/A2A
Step2:
先从Getting Started 开始,它包括了技术文档、SDK、Demo 在这里插入图片描述
Step3:
Agent Card 是什么 ? 可以理解为是 Agent 的名片。也就是说一个 Agent 想要让另一个 Agent 了解自己的名称、能力等,就需要在名片上写清楚。
在这里插入图片描述

Step4:
Task 是什么 ?可以理解为是一间洽谈室,由乙方(发起调用请求的 Agent)邀请甲方(接收调用请求的 Agent)进行会晤。但是会晤的结果(状态)是什么,是甲方立马执行,还是拒绝,还是安排到以后执行等等,这些细节都是由甲方说了算的。
文档给出了 Task 的所有状态集合。至于前面所说的,乙方如何邀请甲方等等动作,则是一个个的接口,需要我们继承实现。
在这里插入图片描述

Step5:
示例代码在 samples 文件夹中,有好几种脚手架实现版本。我们来阅读的 LangGraph 版本吧。该sample是通过 A2A 协议公开的货币兑换,支持多轮对话和流式响应。
在这里插入图片描述

LangGraph 是一个让 AI 智能体“像人类一样思考和工作”的图结构框架,适用于构建复杂、可控、可扩展的 AI 应用系统。

可以看到 A2A 实际上类似 MCP,也是一个 C-S 架构。如果想通过 A2A 去调用另一个 Agent,则首先需要实现一个 A2A 客户端。而被调用的 Agent 呢,则需要实现一个 A2A Server,才能与客户端建立连接。整个消息流转的流程是这样的。先由 A2A Client 向 A2A Server 发送消息,比如用户的 query。Server 收到消息后就会发送给 Agent,由 Agent 执行选择工具,调用工具的流程。之后将工具执行结果返给 Server,由 Server 再返给 Client。

Step6:
看完了原理图,我们再去简单浏览一下代码,印证一下是不是这个流程。先看示例代码的目录结构:
在这里插入图片描述
common: 是 A2A 协议的 SDK
首先是 langgraph 目录__main__.py

import logging
import os
import sys

import click
import httpx
import uvicorn

from a2a.server.apps import A2AStarletteApplication
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.tasks import (
    BasePushNotificationSender,
    InMemoryPushNotificationConfigStore,
    InMemoryTaskStore,
)
from a2a.types import (
    AgentCapabilities,
    AgentCard,
    AgentSkill,
)
from dotenv import load_dotenv

from app.agent import CurrencyAgent
from app.agent_executor import CurrencyAgentExecutor


load_dotenv()

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class MissingAPIKeyError(Exception):
    """Exception for missing API key."""


@click.command()
@click.option('--host', 'host', default='localhost')
@click.option('--port', 'port', default=10000)
def main(host, port):
    """Starts the Currency Agent server."""
    try:
        if os.getenv('model_source', 'google') == 'google':
            if not os.getenv('GOOGLE_API_KEY'):
                raise MissingAPIKeyError(
                    'GOOGLE_API_KEY environment variable not set.'
                )
        else:
            if not os.getenv('TOOL_LLM_URL'):
                raise MissingAPIKeyError(
                    'TOOL_LLM_URL environment variable not set.'
                )
            if not os.getenv('TOOL_LLM_NAME'):
                raise MissingAPIKeyError(
                    'TOOL_LLM_NAME environment not variable not set.'
                )

        capabilities = AgentCapabilities(streaming=True, push_notifications=True)
        skill = AgentSkill(
            id='convert_currency',
            name='Currency Exchange Rates Tool',
            description='Helps with exchange values between various currencies',
            tags=['currency conversion', 'currency exchange'],
            examples=['What is exchange rate between USD and GBP?'],
        )
        agent_card = AgentCard(
            name='Currency Agent',
            description='Helps with exchange rates for currencies',
            url=f'http://{host}:{port}/',
            version='1.0.0',
            default_input_modes=CurrencyAgent.SUPPORTED_CONTENT_TYPES,
            default_output_modes=CurrencyAgent.SUPPORTED_CONTENT_TYPES,
            capabilities=capabilities,
            skills=[skill],
        )


        # --8<-- [start:DefaultRequestHandler]
        httpx_client = httpx.AsyncClient()
        push_config_store = InMemoryPushNotificationConfigStore()
        push_sender = BasePushNotificationSender(httpx_client=httpx_client,
                        config_store=push_config_store)
        request_handler = DefaultRequestHandler(
            agent_executor=CurrencyAgentExecutor(),
            task_store=InMemoryTaskStore(),
            push_config_store=push_config_store,
            push_sender= push_sender
        )
        server = A2AStarletteApplication(
            agent_card=agent_card, http_handler=request_handler
        )

        uvicorn.run(server.build(), host=host, port=port)
        # --8<-- [end:DefaultRequestHandler]

    except MissingAPIKeyError as e:
        logger.error(f'Error: {e}')
        sys.exit(1)
    except Exception as e:
        logger.error(f'An error occurred during server startup: {e}')
        sys.exit(1)


if __name__ == '__main__':
    main()

main 方法的主要工作,是在 66 ~ 74 行代码填充了 AgentCard,然后在第 89 ~ 91 行代码启动了一个 A2AServer。

3. 总结

1)A2A协议是一个C-S架构的HTTP协议,通过在Agent端使用HTTP Server封装,然后在请求端通过规定的协议进行HTTP Client请求来实现Agent之间的通信。

2)A2A协议与MCP是不同层面的协议,解决的问题也不同,可以互补。MCP解决的是如何标准化封装与发布工具的问题,而A2A解决的是Agent间互相通信,形成多Agent的问题,比MCP的维度更高。

3) A2A协议的实现涉及A2A Server和A2A Client端的代码,其中A2A Server端的实现需要根据Agent的实际情况进行描述,而A2A Client端的代码相对简单,主要是获取Agent的名片并发送消息给服务端Agent。

Logo

火山引擎开发者社区是火山引擎打造的AI技术生态平台,聚焦Agent与大模型开发,提供豆包系列模型(图像/视频/视觉)、智能分析与会话工具,并配套评测集、动手实验室及行业案例库。社区通过技术沙龙、挑战赛等活动促进开发者成长,新用户可领50万Tokens权益,助力构建智能应用。

更多推荐