a2a(agent to agent)是一套通信协议，或者说是一套标准化的实现方式。

注：a2a用的是json-rpc协议，这点要特别注意下。

机制(工作原理)

机制图

步骤 1: 能力发现
┌──────────┐                          ┌──────────┐
│  Client  │  GET /.well-known/       │  Agent   │
│          │  a2a-agent-card.json     │          │
│          │ ───────────────────────► │          │
│          │ ◄─────────────────────── │          │
│          │   Agent Card (JSON)      │          │
└──────────┘                          └──────────┘

步骤 2: 发送消息
┌──────────┐                          ┌──────────┐
│  Client  │  POST /a2a/jsonrpc       │  Agent   │
│          │  {method: "message/send"}│          │
│          │ ───────────────────────► │          │
│          │                          │ 处理消息  │
│          │ ◄─────────────────────── │          │
│          │   StreamResponse (流式)   │          │
└──────────┘                          └──────────┘

步骤 3: 任务跟踪（可选）
┌──────────┐                          ┌──────────┐
│  Client  │  GET /tasks/{task_id}    │  Agent   │
│          │ ───────────────────────► │          │
│          │ ◄─────────────────────── │          │
│          │   Task 状态信息           │          │
└──────────┘                          └──────────┘

示例步骤

还是基于pycharm，解释器、虚拟环境等都已配置好。
注：代码不要从网上直接生成，坑很多(主要是版本问题)，建议从git上找示例代码，交给大模型，然后再生成就靠谱多了。

1、安装依赖

pip install a2a-sdk[all] uvicorn fastapi

a2a-sdk这里用的是1.1.0版本(最新版本)。

2、创建simple_agent.py服务端代码

# simple_agent.py - 最简单的 A2A Agent 示例
import asyncio
import uvicorn
from fastapi import FastAPI

from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.routes import create_agent_card_routes, create_jsonrpc_routes, create_rest_routes
from a2a.server.tasks import InMemoryTaskStore
from a2a.types import (
    AgentCard,
    AgentInterface,
    AgentProvider,
    Message,
    Part,
    Role,
)


class SimpleAgent(AgentExecutor):
    """最简单的 Agent 实现"""

    async def execute(self, context: RequestContext, event_queue: EventQueue):
        # 获取用户输入
        user_input = context.get_user_input()
        print(f"收到消息: {user_input}")

        # 创建回复消息
        reply = Message(
            message_id=str(id(context)),
            role=Role.ROLE_AGENT,
            parts=[Part(text=f"Echo: {user_input}")]
        )

        # 发送回复
        await event_queue.enqueue_event(reply)

    async def cancel(self, context: RequestContext, event_queue: EventQueue):
        """取消任务（简单实现）"""
        pass


agent_card = AgentCard(
    name='Simple Echo Agent',
    description='最简单的 A2A 复读机示例',
    provider=AgentProvider(organization='Demo', url='http://localhost:41241'),
    version='1.0.0',
    supported_interfaces=[
        AgentInterface(
            protocol_binding='JSONRPC',
            protocol_version='1.0',
            url='http://127.0.0.1:41241/a2a/jsonrpc',
        ),
        AgentInterface(
            protocol_binding='HTTP+JSON',
            protocol_version='1.0',
            url='http://127.0.0.1:41241/a2a/rest',
        ),
    ]
)
# 创建任务存储和请求处理器
task_store = InMemoryTaskStore()
request_handler = DefaultRequestHandler(
    agent_executor=SimpleAgent(),
    task_store=task_store,
    agent_card=agent_card,
)

app = FastAPI()
app.routes.extend(create_agent_card_routes(agent_card))
app.routes.extend(create_jsonrpc_routes(request_handler, rpc_url='/a2a/jsonrpc'))
app.routes.extend(create_rest_routes(request_handler))

if __name__ == '__main__':
    print("启动 Simple Agent...")
    uvicorn.run(app, host='127.0.0.1', port=41241)

3、创建simple_client.py客户端代码

import asyncio
import httpx
import uuid
from a2a.client import A2ACardResolver, create_client
from a2a.types import Message, Part, Role, SendMessageRequest


async def main():
    # 1. 指定 Agent 地址
    agent_url = "http://127.0.0.1:41241"

    # 2. 获取 Agent 的名片 (自动发现能力)
    async with httpx.AsyncClient() as client:
        resolver = A2ACardResolver(client, agent_url)
        card = await resolver.get_agent_card()
        print(f"已连接到: {card.name} ({card.description})")

        # 3. 创建客户端连接
        agent = await create_client(card)

        # 4. 发送消息并等待回复
        user_input = input("你: ")

        # 构造 Message 对象
        message = Message(
            role=Role.ROLE_USER,
            message_id=str(uuid.uuid4()),
            parts=[Part(text=user_input)],
        )

        # 使用 SendMessageRequest 包装
        request = SendMessageRequest(message=message)

        # 5. 异步迭代响应流
        async for response in agent.send_message(request):
            # 打印回复内容
            if hasattr(response, 'message') and response.message:
                for part in response.message.parts:
                    if part.text:
                        print(f"Agent: {part.text}")
            elif hasattr(response, 'task'):
                print(f"任务状态: {response.task.status}")


# 运行客户端
if __name__ == '__main__':
    asyncio.run(main())

4、python simple_agent.py运行服务端
5、python simple_client.py运行客户端，输入hello，打印出hello即表示成功。

function calling、MCP、a2a的区别?

这是一个经典问题。

比较维度	Function Calling (函数调用)	MCP (模型上下文协议)	A2A (代理到代理协议)
核心定位	AI 模型与外部工具之间的基础交互	AI 模型与外部工具/数据源的统一标准接口	不同 AI 智能体之间的通信与协作标准
交互主体	模型 ↔ 工具 (单点)	模型 ↔ 多工具/数据源	Agent ↔ Agent
解决的问题	让 AI “会动手”，解决单一任务	解决 M×N 集成问题，让 AI “统筹工具”	解决跨框架/跨平台协作，让 AI “一起做大事”
技术机制	模型输出结构化 JSON，由程序执行	客户端-服务器架构，JSON-RPC 通信	基于 Agent Card 发现，基于消息的任务协作
适用场景	查天气、搜索数据库等简单直接的操作	企业级复杂应用、插件生态、实时数据访问	跨平台任务安排、物流管理、多智能体业务流程

a2a比MCP进步在哪里呢?

mcp是精准调用方法，a2a是自然语言，这样有个好处，如果某个业务比较复杂，例如帮我订个机票，需要先查询、再比对、再下单等等。如果调用mcp组合起来很复杂，但是如果用a2a只用告诉它我需要订张机票，它内部如何操作的根本不用管，更加智能了。

相当于只关注于需求和结果，更加智能而且黑盒相对安全。

a2a接口地址大全

1、智能体发现接口（HTTPGET）

这是A2A协议中最具代表性的固定路径，用于客户端在通信前获取目标智能体的“名片”（AgentCard），以了解其能力、支持的输入/输出格式及安全要求。
接口地址：GET/.well-known/agent.json
说明：任何符合A2A协议的客户端都可以通过访问智能体端点URL+/.well-known/agent.json来自动发现该智能体。

2、核心通信接口（JSON-RPC2.0）

A2A的复杂交互（如任务委派、状态追踪）均通过JSON-RPC方法在特定的HTTP端点上完成。根据传输绑定的不同，主要有以下两种端点形式：

①HTTP+JSON传输端点
接口地址：通常为/a2a/{agent-name}或/message（例如/a2a/weather-agent）。
说明：用于标准的HTTP请求和SSE（Server-SentEvents）流式传输。客户端向此端点发送JSON-RPC请求，服务端通过此端点返回结果或推送状态。

②JSON-RPC传输端点
接口地址：通常为/a2a/{agent-name}（与HTTP+JSON可以复用相同路径，也可以配置不同路径）。
说明：专门用于处理JSON-RPC2.0协议请求。服务端可以同时映射这两种绑定，允许客户端根据自身需求选择首选的传输方式。

3、任务与消息操作方法（JSON-RPCMethod）

在上述通信端点上，客户端通过method字段来区分具体的操作意图。以下是核心的方法名及其对应的业务逻辑：

SendMessage：向远程Agent发送消息，用于创建新任务或向现有任务追加信息。
SendStreamingMessage：发送流式消息，允许客户端实时接收任务的状态更新和增量内容。
GetTask：获取指定任务的最新状态和结果（Artifacts）。
ListTasks：按条件列出任务列表。
CancelTask：取消一个正在进行中的任务。
SubscribeToTask：订阅任务更新，通常结合SSE进行重连和实时状态推送。

4、推送通知配置方法

针对耗时较长、客户端可能断开连接的异步任务，A2A提供了Webhook推送通知的配置接口：
CreateTaskPushNotificationConfig：创建推送通知配置。
GetTaskPushNotificationConfig：获取推送通知配置。
ListTaskPushNotificationConfigs：列出推送通知配置。
DeleteTaskPushNotificationConfig：删除推送通知配置。

5、扩展能力获取方法

GetExtendedAgentCard：获取认证后的扩展AgentCard，以便客户端在通过安全验证后，了解远程智能体的完整能力。

总结：
A2A的接口体系非常清晰：对外通过/.well-known/agent.json暴露身份和能力；对内通过/a2a/{agent-name}等端点接收JSON-RPC请求，利用SendMessage、GetTask等方法管理任务的完整生命周期。

要不要我帮你整理一份A2A和MCP的对比速查表？方便你在实际项目中快速判断什么场景该用哪个。

其他

文档

a2a官方github地址(实测可以访问，看网络情况)：
https://github.com/google/A2A