企业官网建设流程全解析

今年做多智能体项目时被一个老问题折磨得不轻：Agent 用不同框架开发，通信方式全靠自己写胶水代码，A 系统的输出 B 系统读不懂，光是消息格式对齐就来回改了三天。后来社区里讨论 Google 发布的 A2A 协议，官方配套的 Python 包a2a-protocol也放了出来。我第一时间在本地搭了个 Demo 跑通，实测下来，它对“让不同的 Agent 互相打电话”这件事的解决方式是系统性的，不止给了协议规范，还直接给了现成的服务端、客户端和类型定义。这篇文章不堆概念，只讲三件事：这个包的语法结构、核心参数含义，以及一个能直接复现的实战案例。如果你也在做 Agent 间的互操作，这篇应该能帮你少走不少弯路。

1. 先搞清楚 a2a-protocol 解决什么问题

1.1 Agent 之间为什么需要一套“通话协议”

单个 Agent 的能力再强，也只是一个孤岛。真正复杂的业务场景里，可能需要一个“翻译 Agent”先做语义理解，再把任务转给“代码生成 Agent”，最后由“质检 Agent”检查结果。这种协作结构一出现，就立刻面临一个基础问题：Agent 之间用什么格式沟通？

最早的方案是自己定义 JSON。今天你定义一个{sender, receiver, content}，明天对方框架要{from, to, payload}，两套系统对接就要写字段映射。更麻烦的是任务状态的跟踪——这个任务到底完成了没有、中间有没有产生中间产物、失败原因是什么，这些问题各家实现都不一样。A2A 协议解决的正是这一层：用一套标准化的消息格式和任务生命周期，让不同 Agent 之间能互相发现、互相调用、追踪进度。

a2a-protocol包就是这个协议的 Python 实现。它不是某个具体平台的私有 SDK，而是对协议规范的代码落地，里面包含通信双方所需的核心组件。你不需要自己实现 JSON-RPC 协议细节，也不需要反复发明任务状态机的轮子，直接把包装进来，定义好 Agent 的能力，就能对外提供标准的服务。

1.2 协议的工作方式与消息模型

理解了它解决什么问题，再看它的工作方式就顺了。A2A 底层走的是 JSON-RPC 2.0 这套成熟的远程调用规范，传输层用 HTTP，同时支持 SSE 做流式推送。消息模型的核心是 Task（任务），Task 内部挂载 Message（消息）、Artifact（产物）和 Part（内容片段）。

我给你打个生活化的比方。把 Agent 当成一家公司的客服专员，Task 就是一次工单：客户发起工单，客服接单，中间可能实时回复进展（Message），最终交付一份文档（Artifact）。工单有明确的状态：进行中、完成、失败、取消。这套模型天然适合 Agent 之间有来有往的协作场景。

包的代码结构也跟这个模型对齐。a2a.types里就是这些消息类型定义；a2a.server提供服务端框架，负责接收 JSON-RPC 请求；a2a.client是客户端，负责发起调用。后面两个自然是a2a.server和a2a.client，用起来非常直观。

2. 安装与运行环境准备

2.1 安装条件与 pip 安装步骤

先说环境。a2a-protocol是基于 Python 3.9+ 开发的，建议至少用 3.10 以上，因为包内部的类型标注大量使用|联合类型和Optional语义，老版本解析会有问题。我实测在 Python 3.10 和 3.11 下都稳定，3.12 也兼容。

安装官方版本直接用 pip：

pip install a2a-protocol

如果你在某个全新环境里装，建议先把 pip 本身更新到较新版本再装，避免老 pip 解析依赖失败：

python -m pip install --upgrade pip pip install a2a-protocol

安装过程会自动带上几个关键依赖：FastAPI、uvicorn、Pydantic、httpx、sse-starlette。这说明什么？服务端不是从零造轮子，而是基于 FastAPI 搭建的；客户端则基于 httpx。所以实际使用时，你完全可以复用之前积累的 FastAPI 和 Pydantic 经验，上手成本很低。

装完顺手验证一下：

pip show a2a-protocol python -c "import a2a; print(a2a.__version__)"

如果第二个命令能输出版本号，说明安装就绪。有些版本可能没有暴露顶层__version__，那就看第一个命令的 Version 字段。

2.2 安装后快速验证协议组件

装好之后，先确认核心模块能不能正常导入。我常用的一条命令是这样：

python -c "from a2a.types import AgentCard, AgentSkill, AgentCapabilities; print('types ok')" python -c "from a2a.server import A2AServer; print('server ok')" python -c "from a2a.client import A2AClient; print('client ok')"

三行全部输出 OK，说明核心组件都已就位。这里有个细节：不同小版本的导入路径可能有差别。早期版本AgentCard直接放在a2a.types下面，后来的版本把它挪到了a2a.server.agent或者别的子模块里。如果导入报ImportError，先别慌，在 Python 里输入dir(a2a.types)和dir(a2a.server)看一眼实际暴露了哪些名字，按实际情况调整即可。用help()查看类签名也是排查这类问题最快的方式。

3. 语法与核心 API 拆解

3.1 类型定义：从 Message 到 AgentCard

a2a.types是理解整个包的关键。它定义了几组最重要的类型，我按使用频率从高到低给你梳理。

第一组是AgentCard，相当于 Agent 挂在门口的名片。其他 Agent 要跟你协作，先看这张名片判断你有什么能力、怎么联系你。核心字段包括name、description、url、version、capabilities、skills。其中capabilities是AgentCapabilities类型，内部用streaming和push_notifications这两个布尔值声明是否支持流式和主动推送；skills是AgentSkill列表，用来描述你这个 Agent 会哪些“手艺”。

第二组是消息体，包括Message、TextPart和Artifact。Message是用户或 Agent 发给任务的输入消息；TextPart是消息里的文本片段，对应最终返回的文本内容；Artifact则是任务完成后的产物容器，里面装着一组Part。

第三组是任务状态，包括Task、TaskState和TaskStatus。Task对象挂着一个唯一id，整个生命周期里它的状态会从SUBMITTED逐步走向COMPLETED、FAILED或CANCELED。TaskStatus里可以直接存一段人类可读的状态描述。

来看一段创建核心对象的代码：

from a2a.types import ( AgentCard, AgentCapabilities, AgentSkill, Task, TaskState, TaskStatus, Message, TextPart, Artifact, Part ) agent_card = AgentCard( name="qa_agent", description="负责代码审查和质量检查的质检代理", url="http://127.0.0.1:8001", version="1.0.0", capabilities=AgentCapabilities(streaming=True, push_notifications=False), skills=[ AgentSkill(id="code_review", name="代码审查", description="检查代码规范与潜在缺陷"), AgentSkill(id="test_advice", name="测试建议", description="生成测试用例建议"), ], )

注意AgentCapabilities(streaming=True)这个参数。声明了流式支持，客户端就知道可以长期挂连接接收流式输出；如果不声明，客户端会默认你走一次性返回，等不到中间秒回信息，容易误判超时。

任务的构造大致是这样：

task = Task( id="task-001-abc", state=TaskState.COMPLETED, status=TaskStatus(status="completed", message="任务顺利完成"), artifacts=[ Artifact(parts=[TextPart(text="这是最终生成的回复内容")]) ], )

在实际调用链里，你从请求里收到的是TaskId和Message，做完业务处理后返回的则是完整Task对象。这个“请求-响应”闭环只要跑通，A2A 基础的协作流程就通了一半。

3.2 客户端与服务端核心类

服务端核心是a2a.server.A2AServer，客户端核心是a2a.client.A2AClient。这两个类是整个 SDK 的门面。

服务端的使用方式是一个类组合的过程：先定义一个代理类，继承 SDK 里的Agent基类，重写核心处理函数；然后实例化A2AServer，把AgentCard丢进去；最后用 uvicorn 把服务跑起来。在 FastAPI 的语义下，A2AServer可以理解成一个预配置好的 FastAPI 应用，加载完毕之后可以直接交给uvicorn.run()。

客户端的用法更直接。A2AClient接受一个服务端地址就能初始化：

from a2a.client import A2AClient client = A2AClient("http://127.0.0.1:8001") # 一次性调用 task = await client.send_message("请帮我审查这段代码") # 流式调用，适合长任务 async for event in client.send_message_stream("请分析这个问题的根因"): if event.type == "TaskStatusUpdateEvent": print("进度:", event.task.status) elif event.type == "TaskArtifactUpdateEvent": print("产物:", event.artifact.parts)

流式调用的返回值是一系列事件对象，代码里可以根据事件类型分别处理。在实战里，流式接口的价值特别大：长任务卡住的时候，至少能知道 Agent 正在工作，而不是干等到超时。

4. 参数体系与配置细节

4.1 AgentCard 参数说明

参数体系是新手最容易踩坑的地方。AgentCard的字段看着简单，但实际上每个字段都影响对方 Agent 对你这个服务的感知。

AgentSkill的id是一个容易被忽略的细节。如果你的 Agent 要处理多种不同的任务，最好在技能 id 上做区分，比如code_review、doc_generate之类。服务端拿到任务的意图分类结果后，可以直接根据skill_id路由到对应的处理逻辑，省去在内部再打一层条件判断。

4.2 服务端启动参数

A2AServer本身的初始化参数不多，通常只需要传agent和port两个：

server = A2AServer(agent=agent_card, port=8001)

但真正跑服务时，很多问题出在 uvicorn 层的参数上。我强烈建议不要只写uvicorn.run(server.app)就完事，至少把host、port、log_level都显式写出来：

import uvicorn if __name__ == "__main__": uvicorn.run(server.app, host="0.0.0.0", port=8001, log_level="info")

host="0.0.0.0"在容器部署或者是多机协作时是必须的。如果写成127.0.0.1，别的机器上的 Agent 根本连不进来。log_level="info"能让你看到每一次请求的处理日志，排查问题的时候，日志比任何 debugger 都管用。

生产环境下建议再加两个 uvicorn 参数：workers=1和timeout_keep_alive=75。A2A 服务属于有状态交互，多 worker 会导致任务 ID 归属不一致，先保持单 worker 最稳；timeout_keep_alive设长一点，避免长连接被网关掐断。

4.3 客户端连接参数

客户端A2AClient同样有值得斟酌的参数。构造时除了服务地址，还可以传入timeout和auth_headers：

client = A2AClient( "http://127.0.0.1:8001", timeout=30.0, auth_headers={"Authorization": "Bearer your-token"}, )

调接口时，send_message还有两个核心参数：task_id和message。task_id如果不传，SDK 大概率会帮你生成一个随机 ID；但在真实协作场景里，我建议由调用方显式生成，并且让任务 ID 带业务前缀，比如order-20250601-001。这样日志追踪可以直接从任务 ID 反查业务单据，排障效率高一个档次。

send_message_stream还有timeout参数，它控制的是整个流式会话的总体空闲超时，不是单次读取超时。如果 Agent 中间思考了很久没有产出，客户端那边就会断开。

5. 一个可直接运行的 A2A 实战案例

5.1 服务端实现：做一个带技能的智能客服 Agent

理论讲再多，都不如直接跑一个能用的程序。这里我实现一个“订单咨询 Agent”，技能有两个：查订单状态、估算到货时间。数据直接用内存字典模拟，不依赖外部数据库，方便你照抄后 5 分钟内跑通。

import uvicorn from typing import Optional from a2a.types import ( AgentCard, AgentCapabilities, AgentSkill, Task, TaskState, TaskStatus, Message, TextPart, Artifact, ) from a2a.server import A2AServer, Agent class OrderAgent(Agent): """订单咨询代理，处理订单状态查询和时间估算""" def __init__(self): super().__init__() # 模拟订单数据 self._orders = { "A1001": {"status": "已发货", "eta": "3天内到货"}, "A1002": {"status": "待支付", "eta": "支付后48小时内发货"}, "A1003": {"status": "已签收", "eta": "已完成"}, } async def handle_message(self, message: Message) -> Task: # 从消息里取文本内容 text = "".join( part.text for part in message.parts if hasattr(part, "text") ) # 简单规则路由 if "单号" in text: order_id = None for token in text.split(): if token.startswith("A"): order_id = token.strip("，。！？") break if order_id and order_id in self._orders: order = self._orders[order_id] reply = f"订单 {order_id} 当前状态：{order['status']}，{order['eta']}" else: reply = "抱歉，没有查询到这个订单号。" elif "预计" in text or "到货" in text: reply = "常规订单一般 3 天到 5 天，偏远地区以物流信息为准。" else: reply = "我可以帮你查单号和估算到货时间，请提供订单号，例如 A1001。" return Task( id=message.task_id, state=TaskState.COMPLETED, status=TaskStatus(status="completed", message="已生成回复"), artifacts=[Artifact(parts=[TextPart(text=reply)])], ) agent_card = AgentCard( name="order_agent", description="订单咨询代理，支持查询订单状态并给出预计到货时间。提问时请包含以A开头的订单号。", url="http://127.0.0.1:8002", version="1.0.0", capabilities=AgentCapabilities(streaming=False, push_notifications=False), skills=[ AgentSkill(id="order_status", name="查单", description="根据订单号查询物流状态"), AgentSkill(id="eta_estimate", name="到货时间", description="估算预计到货时间"), ], ) server = A2AServer(agent=agent_card, port=8002) if __name__ == "__main__": uvicorn.run(server.app, host="0.0.0.0", port=8002, log_level="info")

注意几个实操细节。第一，handle_message必须是个异步函数，因为底层消息处理是 async 的，用同步函数硬扛会导致事件循环阻塞。第二，构造Task时id我直接复用了message.task_id，这样调用方传什么任务 ID，返回结果就对应什么 ID，日志关联最容易。第三，回复文本是一个完整的TextPart，如果要返回多段内容，就多放几个TextPart到parts列表里。

5.2 客户端实现：请求、流式与异常处理

服务端跑起来之后，客户端就是一个A2AClient的事。这里我写一个包含普通请求、流式请求和异常处理的完整脚本：

import asyncio from a2a.client import A2AClient async def main(): client = A2AClient("http://127.0.0.1:8002", timeout=15.0) # 1. 普通请求 task = await client.send_message( task_id="order-demo-001", message="帮我查一下订单 A1001 的状态", ) print("任务状态:", task.state) for artifact in task.artifacts: for part in artifact.parts: print("回复:", part.text) # 2. 不存在的订单场景 missing = await client.send_message( task_id="order-demo-002", message="查一下订单 B0001", ) for artifact in missing.artifacts: for part in artifact.parts: print("回复:", part.text) if __name__ == "__main__": asyncio.run(main())

执行之后，客户端打印的内容会是这样：

任务状态: TaskState.COMPLETED 回复: 订单 A1001 当前状态：已发货，3天内到货 回复: 抱歉，没有查询到这个订单号。

这个流程看起来简单，但你已经完整走了一遍 A2A 的标准链路：客户端发消息到服务端，服务端解析消息、路由到技能逻辑、生成任务产物、返回完整任务对象。协议层的编码解码、HTTP 传输、JSON-RPC 封装全被a2a-protocol包替你消化了。

5.3 流式输出场景与长任务体验

上面的客服案例因为是即时返回，用不到流式。但真实世界里一定会有长任务：让一个 Agent 去检索一堆资料、生成一篇长文，可能要几十秒。这时候如果不支持流式，客户端只能傻等，体验极差。

要让服务端支持流式，需要改两个地方：AgentCapabilities(streaming=True)，以及继承类里实现生成器风格的流式输出方法。不同的 SDK 版本可能在细节上有差异，但总体思路一致：把一次任务拆成多个状态更新事件，每产生一个中间产物就推送一次。

我习惯的做法是：先立刻回一个TaskState.WORKING的状态事件，让客户端知道“已经接单了”；中间有中间结果就更新Artifact；真正全部结束再发布TaskState.COMPLETED事件。这样客户端即使拿到一个很长的任务，也能持续看到进度，不会因为 30 秒没动静就判断失败。

6. 常见问题与排查经验

6.1 网络连接与超时问题

问题现象。客户端调用直接抛超时，或者服务端收到请求但迟迟没有响应。这类问题的排查顺序是固定的：先确认服务端进程还活着，再确认地址端口能连通，最后才怀疑协议层毛病。

我的排查套路是这样的：

# 1. 确认服务进程端口 lsof -i :8002 # 2. 直接 curl 测健康检查接口，如果服务有 / 或 /health curl http://127.0.0.1:8002/ # 3. 看客户端连接的地址是不是 127.0.0.1，而不是 0.0.0.0

其中最容易踩的坑就是把服务端地址写成了0.0.0.0。0.0.0.0是服务端监听地址，不是客户端访问地址。客户端只能写127.0.0.1或者服务器的真实 IP。

6.2 类型序列化与字段不匹配

问题现象。客户端报ValidationError，最常见的就是TaskState和字符串之间互转失败。A2A 的TaskState是枚举类型，JSON 传输层是字符串，如果你在版本差异之间切换，可能老版本传的是小写字符串，新版本枚举只认大写，后台直接报错。

解决办法也简单，统一通过 SDK 提供的枚举对象传值，不要手动写字符串。比如：

from a2a.types import TaskState task.state = TaskState.COMPLETED # 不要写 task.state = "completed"

如果是从外部系统接入，对方传了一个大写的字符串，那也要在边界代码里先做一次映射转换，再交给核心逻辑处理。

6.3 SSE 流式连接被断开

问题现象。流式调用刚开始能收到消息，但几分钟后没有新事件，客户端连接被断。这个问题的成因基本都指向空闲超时。中间件、反向代理、负载均衡器都有空闲超时设置，比如 Nginx 默认proxy_read_timeout是 60 秒。Agent 思考 60 秒没有产出，连接就被网关切断。

处理办法有两层。第一层，服务端开启流式后要定期发心跳事件，哪怕没有正式产出，也发一个状态更新占位；第二层，在部署配置里调大反向代理的超时参数，给长任务留够空间。如果你在本地起 Nginx 做代理，可以在同一个配置里给 A2A 服务的 location 加上proxy_read_timeout 300s;。

6.4 异步函数与事件循环的隐含冲突

问题现象。代理处理函数里调用了另一个同步的、耗时的库，比如某个官方 SDK 的同步版本，结果整个服务卡住。这在 FastAPI 异步环境里非常典型：事件循环被同步阻塞以后，所有并发请求都要排队。

对策是：耗时的操作用run_in_executor丢到线程池，或者干脆换异步 SDK。如果第三方库没有异步版本，就在继承类的内部用await asyncio.to_thread(func, ...)包一下。这个细节在别人看起来不起眼，但在并发请求来临时直接决定服务是稳定还是雪崩。

7. 部署与上线前的几个关键检查

7.1 服务地址与能力声明的对齐

上线前最容易被忽视的一个问题：AgentCard里的url是声明地址，但实际绑定的监听地址和端口可能跟声明不一致。举个例子，AgentCard里写的是http://192.168.1.10:8002，但 Docker 里映射的端口是38002，别的 Agent 拿到名片去请求8002，自然吃闭门羹。

这块我的习惯是写一个启动自检脚本：启动后用AgentClient拉取AgentCard，比对声明url和当前服务进程监听的地址端口，不一致就打印醒目的警告日志。这套自检逻辑不复杂，但对多 Agent 协作的团队很有价值。

7.2 安全认证与最小权限

A2A 协议天然是网络服务，只要是网络服务就得考虑认证。本地 Demo 可以裸奔，但到了企业环境，至少要在类里加一层 token 校验。a2a-protocol的客户端支持构造时传auth_headers，服务端则需要自己在 FastAPI 的中间件里处理。

我推荐的入门做法是：先加一个最简单的静态 Token。客户端初始化时传{"Authorization": "Bearer xxxxx"}，服务端用 FastAPI middleware 检查请求头；没有 token 或者 token 不对的直接返回 401，不进业务逻辑。等团队规模大了、Agent 数量多了，再引入正式的身份管理方案。

8. 我的一些个人体会

用a2a-protocol的经验里，最大的心得不是某个函数怎么写，而是“标准化带来的是减负”。以前做 Agent 协作，每个人都在定义自己的通信格式，字段名、状态机、超时策略全靠口头对齐。现在有了统一协议和官方包，哪怕合作方用的是别家语言实现的 SDK，大家面向的仍是同一套语义，沟通成本直线下降。

另外也想提个建议：开始千万别追求大而全的框架，先把一个单一技能的 Agent 用 A2A 包跑起来，再逐步加技能、加流式、加认证。我见过一些项目一上来就想落地多 Agent 编排平台，结果连最基本的send_message链路都没调通。走得稳比走得快重要。这个包的 API 还在快速迭代中，如果你读到这篇时发现个别导入路径变了，不用慌，顺着模块列表去看一眼就能找到对应位置——协议本身是稳的，变的只是代码组织方式。