企业官网建设流程全解析

上篇我们聊了大方向上的Agent协作形态，为什么需要Agent之间直接对话，以及A2A协议解决的核心问题。这篇进入“真刀真枪”的部分：消息到底长什么样、连接怎么建立、任务状态怎么流转、鉴权怎么做，最后给你一个能在本地跑起来的最小Demo，再加上我联调过程中踩过的几个坑。如果你是那种“看完协议文档总觉得差点意思”的人，这篇应该能补上那段距离。

1. 从Agent Card开始：连接建立前的那张“名片”

A2A协议里有个很朴素的前提——一个Agent要能主动发起任务，它得先知道另一个Agent在哪儿、能干什么、怎么联系。这跟两个人合作一样，你可以不认识对方，但总得先拿到对方的名片。Agent Card就是这个名片。

1.1 一张Agent Card里到底写了什么

按照A2A协议规范的思路，Agent Card是一个JSON文档，描述Agent的公开信息。我在自己的模拟项目里维护的卡片大概长这样：

{ "name": "Zenith Translation Agent", "description": "负责英文到中文的文本翻译，支持技术文档、日常文本风格", "version": "1.2.0", "url": "https://agent.example.com/translation", "endpoints": [ { "path": "/a2a", "protocol": "a2a", "transport": "http" } ], "capabilities": { "task_state": true, "streaming": true, "push_notifications": false }, "skills": [ { "id": "translate_text", "name": "翻译文本", "description": "输入英文文本，输出中文翻译", "input_modalities": ["text", "file"], "output_modalities": ["text", "file"] }, { "id": "format_doc", "name": "格式化文档", "description": "对原始文档进行结构化整理，输出Markdown格式", "input_modalities": ["text"], "output_modalities": ["text"] } ], "provider": { "organization": "example.dev", "version": "1.0.0", "reference": "agent-card-2024" } }

别被这么多字段唬住。最关键的就三个：

1. endpoints——Agent暴露的调用入口，既包括A2A专用路径，也可以包含HTTP的握手路径。这是别人“敲门”的门牌号。
2. capabilities——Agent支持哪些交互方式，比如是否支持长任务状态查询、是否支持流式返回、是否支持主动推送。
3. skills——Agent能干的活儿，每一项都定义清楚输入输出用文本还是文件。这相当于工具清单，客户端会根据这个清单决定把什么任务派给谁。

我见过不少新手一上来就急着写业务代码，把Agent Card当成“事后补的配置文件”。实际上这个卡片是协作的源头。客户端通常都是先读卡、再决定调不调你，卡片写得不清楚，Agent做得再好也可能没人调。

1.2 卡片的发现与缓存刷新

Agent Card的获取方式一般是直接通过URL拉取。业界常见的做法是：Agent把卡片放在一个稳定路径下，比如https://agent.example.com/.well-known/agent.json或者由注册中心返回。你在本地调试时，也可以把它当成一个静态文件让HTTP服务直接serve出来。

这里有一个非常容易踩的坑：客户端会缓存Agent Card，而且缓存刷新策略各不相同。

我在某个内部模拟项目中遇到过：Agent上线了新skill，我更新了卡片，但另一边的编排程序调用了半天还是提示“unknown skill”。排查到最后才发现，对方把卡片缓存存在本地Redis里，TTL设置成了24小时。我改完卡片，对方要隔一天才拿到最新能力。

如果你在写客户端，建议对Agent Card做两级处理：

• 会话外缓存（比如5-10分钟），降低频繁拉卡的带宽开销；
• 一旦收到agent_card_updated这类变更通知，立即强制刷新。

如果你在写Agent服务端，就要保证卡片接口不要被网关缓存误伤。许多团队在Nginx层给静态JSON做了强制缓存，忘了排除Agent Card路径，结果所有协作方都被旧卡片误导。

2. 消息是JSON-RPC：A2A通信的“包裹”长什么样

A2A没有发明一套新的消息语言，而是直接采用了JSON-RPC 2.0作为消息层基础。这个决策在实践上非常省事：不必自己定义序列化规则和错误格式，所有语言生态里几乎都有现成的JSON-RPC库，或者至少你能很快手写一个解析器。

2.1 为什么要用JSON-RPC而不是REST

有很多人问过我这个问题。我个人的理解是：Agent协作是一种“远程能力调用”，它天然对应RPC模型。你对另一个Agent说“帮我翻译这段文字”“帮我查一下某份文档”，本质上是一个带参数的过程调用。REST风格更适合资源管理，但不太适合这种“请求某个Agent干活”的语义。

JSON-RPC 2.0有几个优势：

• 请求/响应结构非常清楚，字段固定，容易做中间层debug；
• 标准里定义了错误对象（error code），协作双方可以快速对齐失败原因；
• 支持“通知”类消息，即不需要对方返回响应的单向消息，适合做事件通知；
• 与语言和传输层解耦，HTTP、WebSocket、SSE都能承载同一套消息体。

A2A对JSON-RPC做了一些扩展，比如在params里携带任务ID、消息对象、会话上下文等。但核心骨架还是标准的那一套。

2.2 一条消息从客户端到Agent的完整链路

我以一个最简单的“发送任务”请求为例，展示一条消息从调用方到Agent的完整链路。假设编排方要让翻译Agent执行一次翻译：

{ "jsonrpc": "2.0", "id": "req-8f3c8d0e-b2f7-4b1f-9a52-4a2c9c7f8b30", "method": "tasks/send", "params": { "taskId": "task-1701000000001", "message": { "role": "agent", "sender": "agent-orchestrator-01", "time": "2025-05-12T10:00:00Z", "parts": [ { "type": "text", "text": "请把下面这段英文翻译成中文：\nThe quick brown fox jumps over the lazy dog." } ] } } }

逐字段拆解：

• jsonrpc和id是JSON-RPC的固定内容，id由调用方生成，用来关联请求和响应。很多人会忽略id的唯一性，这个问题我后面在踩坑部分专门说。
• method是A2A定义的方法名。常见的包括：
  • tasks/send：向Agent发起一个新的任务，并携带第一条消息；
  • message/send：在已有任务上继续发消息；
  • tasks/get：查询任务状态；
  • tasks/cancel：请求取消任务；
  • tasks/pushNotificationConfig：配置push通知。
• params.taskId：标识一个任务实例。注意任务ID由调用方主动生成，而不是Agent返回——这是很多刚从REST习惯转过来的人需要适应的点。
• params.message：消息本身由role、sender、time和parts组成。parts是消息内容的数组，每个part都有一个type字段，可以是text、file或data。

Agent端收到这条消息后，正常会返回一个tasks/send响应。如果任务是异步的，响应里会带上当前任务状态，一般是working。客户端再通过tasks/get轮询或通过streaming拿后续结果。

2.3 内容对象与artifacts的边界

A2A协议里有两个容易混淆的概念：message.parts和artifacts。

简单来说，parts是“Agent之间的对话内容”，有点像一个Agent对另一个Agent说的一句话，这句话可能是文本、文件引用或者结构化数据。而artifacts是“Agent完成工作后产出的成果物”，是这个Agent“做出来”的东西。

我举个例子。你让翻译Agent“翻译并整理一份技术文档”，你的请求文本是parts；翻译完成后Agent返回的Markdown格式文档，属于artifacts。

这两个概念在协议实现里对应不同的数据结构，千万别混在一起。如果你想让下游Agent直接消费结果，就应该把结果放到artifacts里，让下游用file或data类型引用。如果只是对话中临时传一下，再放parts也不迟。这个区分直接影响协作链路里下游Agent的解析逻辑。

3. 任务生命周期：多Agent协作真正在“跑”的东西

两个Agent之间的一次合作，在A2A协议眼里就是一个“任务”（task）。任务从创建到结束，经历一系列状态。理解这套状态机，基本上就理解了A2A的协作骨架。

3.1 任务状态机：从submitted到completed

在A2A规范草案里，一个任务通常包含以下几种状态：

正常流程下，状态按照submitted -> working -> completed流转。但input-required这个状态特别值得多说两句，因为它在Agent协作场景里非常常见。

3.2 为什么不建议所有链路都做成“一次到位”

很多Agent协作框架喜欢把任务做得“全自动、无人工”。但实际业务场景里，下游Agent经常遇到信息不足的情况。

我遇到过的一个真实例子：一个数据分析Agent要调用一个报表Agent生成月度报告。数据分析Agent只传了“生成5月份报告”这个指令，但没有说明报告模板和面向的受众。报表Agent返回了input-required状态，列出了自己缺失的两项信息。数据分析Agent收到后补充了“面向管理层、使用简化模板”，任务才继续往下走。

A2A把input-required设计成合法状态，就是在承认一个事实：Agent之间的协作不是每次都能靠一个prompt打通。与其强行让Agent猜，不如让下游Agent明确表达“我需要什么”。

在实际工程中，我建议在Agent端显式处理input-required，而不是把它当作异常返回失败。当Agent发现自己拿到的信息不足以完成用户需求时，应该生成一个结构化反馈，列出缺什么、建议怎么补。这样协作链路虽然多了一轮往返，但成功率会明显提高。

3.3 多轮交互：轮询、流式与推送

任务状态有两种获取方式。最简单的是轮询——客户端不停地调tasks/get，直到任务进入终态。代码简单，但在长任务场景下效率不高。

更推荐的是流式。A2A里的streaming通常基于SSE（Server-Sent Events），Agent在处理任务的过程中持续向客户端推送任务状态更新。这种机制非常适合“生成一段吐一段”的场景，比如翻译长文档、逐段总结会议纪要。

第三种是push通知。Agent在任务状态变更时主动给客户端配置好的webhook地址发通知。相比SSE，push适合那些处理时间特别长、客户端不想一直挂着连接的任务。

我给你的选型建议很简单：

• 本地调试、任务秒级返回：轮询就行；
• 任务在十几秒到几分钟，且希望体验流畅：上SSE流式；
• 分钟级以上或跨网调用：用push通知，让Agent在完成时主动告诉你。

尽量不要一上来就把三种方式全实现。先做轮询，跑通了再按需加streaming，最后再考虑push。能力太多会增加联调成本。

4. 鉴权与信任边界：Agent网络里最容易翻车的地方

A2A协议的早期版本设计得很酷，但有一点很容易被忽略——怎么保证只有该调用你的人能调用你，以及Agent之间怎么证明“自己是自己”。

4.1 三种鉴权等级怎么选

A2A协议规范里提到了几种鉴权思路，大致可以分成三类：

在本地跑Demo时，用第一种最省事。但只要你把Agent网络暴露到非完全可信环境，就至少要上Bearer Token。原因很简单：Agent是有执行力的，不是只读接口。别人一旦能调用你的Agent，就是免费利用你的算力、数据、甚至对外发消息的能力。

4.2 用户授权传递与权限最小化

多Agent协作里有一个容易被忽略的细节：如果用户的请求要经过多个Agent，那下游Agent应该以什么身份执行？是用户的身份，还是调用方Agent的身份？

按照我在项目里的经验，这里最好遵循“权限最小化”原则。每个Agent只拥有完成自身任务所需的最小权限，不要把它上游的完整用户context一路透传下去。比如翻译Agent只需要“读取原文、返回译文”的权限，不该拥有用户数据删除权限。

A2A协议允许在消息中携带上下文和用户相关信息，但在实践中我会非常克制。除非确有必要（比如要结合用户偏好做个性化处理），否则一律不传用户敏感信息。Agent链越长，数据暴露面越大，每多一次转发就多一份泄露风险。

我在做模拟项目X时，一开始图省事，让编排Agent把用户原始token直接传给所有下游Agent。后来发现翻译Agent的日志系统会把完整token打出来，当场吓出一身冷汗。从那以后，凡是跨Agent调用，一律只传最小化的用户身份引用，具体敏感凭证由专门服务管理。

4.3 本地联调时怎么处理鉴权

如果你的Agent是本地起服务的，先别急着上OAuth2。我推荐的分步策略是这样的：

1. 本地首版：关闭鉴权，专注把协议流程跑通；
2. 内网测试：上Bearer Token，用环境变量管理密钥，别写死在代码里；
3. 跨网络发布：再考虑完整OAuth2流程，并且认真设计scope。

另外请记住，Token的传输一定要走HTTPS，否则和明文裸奔没区别。本地可以HTTP，远程必须TLS。

5. 一个最小可用的多Agent协作Demo

光讲概念不给代码，总觉得没落地。我这边给你一个可以本地直接跑通的最小Demo：两个Agent服务，A负责接收“翻译+整理”任务，B负责做翻译的核心处理。其实严格来说这个Demo是“编排方-执行Agent-翻译能力”的最小链路，但你完全可以按这个模式扩展成更多Agent的协作网络。

5.1 场景设计

我设计了一个“技术文档翻译整理”场景：

• 一个“编排Agent”（或者叫客户端Agent），它收到用户提交的英文文本，需要把任务转交给翻译Agent；
• 翻译Agent内部是个独立服务，它实现A2A协议接口，对外暴露一个translate_textskill；
• 翻译Agent完成任务后，把结果以artifact的形式返回。

我用Python写，Web框架选择了FastAPI，再用一个极其简单的离线词库映射实现“翻译”。这样你不需要申请任何第三方API key也能跑通。

项目结构长这样：

a2a-demo/ ├── agent_a.py # 翻译Agent服务端 ├── client_b.py # 调用方（编排方） └── requirements.txt

5.2 翻译Agent服务端实现

from fastapi import FastAPI, Request, Response import json import uuid import time app = FastAPI() # 一个简化版“翻译引擎”，实际项目中换成翻译API即可 TRANSLATION_MAP = { "hello": "你好", "world": "世界", "agent": "智能体", "protocol": "协议", "collaboration": "协作" } def simple_translate(text: str) -> str: # 对英文文本做简单的单词映射翻译 words = text.lower().replace(",", "").replace(".", "").split() translated = [TRANSLATION_MAP.get(w, w) for w in words] return " ".join(translated) def build_result(status: str, task_id: str, artifact=None): result = { "jsonrpc": "2.0", "id": str(uuid.uuid4()), "result": { "task": { "id": task_id, "status": status, "history": [ { "role": "agent", "time": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "parts": [] } ] } } } if artifact: task = result["result"]["task"] task["history"][0]["parts"].append({ "type": "file", "name": "translated.txt", "url": "http://localhost:8000/files/translated.txt", "content": artifact }) task["artifacts"] = [ { "name": "translated.txt", "type": "text/plain", "content": artifact } ] return result @app.post("/a2a") async def a2a_endpoint(request: Request): body = await request.json() method = body.get("method") if method == "tasks/send": params = body.get("params", {}) task_id = params.get("taskId", "") message = params.get("message", {}) # 提取消息文本 parts = message.get("parts", []) full_text = "" for part in parts: if part.get("type") == "text": full_text += part.get("text", "") + " " # 模拟内部Agent执行处理 translated = simple_translate(full_text.strip()) # 返回完成状态和结果 return build_result("completed", task_id, artifact=translated) if method == "tasks/get": params = body.get("params", {}) task_id = params.get("taskId", "") return build_result("completed", task_id) return Response( content=json.dumps({"jsonrpc": "2.0", "error": {"code": -32601, "message": "Method not found"}, "id": body.get("id", None)}), media_type="application/json" )

这里我故意没写Cancellation、Streaming和Push，先把最核心的tasks/send和tasks/get跑通。这个服务本质上是把协议层接口暴露在/a2a路径上，内部逻辑就是普通的函数处理。

5.3 调用方（编排方）实现

import requests import json import uuid import time AGENT_URL = "http://localhost:8000/a2a" def send_task(text: str): task_id = f"task-{int(time.time() * 1000)}" payload = { "jsonrpc": "2.0", "id": f"req-{uuid.uuid4()}", "method": "tasks/send", "params": { "taskId": task_id, "message": { "role": "agent", "sender": "agent-orchestrator-01", "time": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "parts": [ { "type": "text", "text": text } ] } } } resp = requests.post(AGENT_URL, json=payload) print("原始响应:", json.dumps(resp.json(), indent=2, ensure_ascii=False)) data = resp.json() if "result" in data: task = data["result"]["task"] print("任务状态:", task["status"]) if task.get("artifacts"): print("翻译结果:", task["artifacts"][0]["content"]) if __name__ == "__main__": send_task("hello agent, welcome to protocol collaboration")

保存后在两个终端分别执行：

python -m uvicorn agent_a:app --port 8000 python client_b.py

如果能看到类似这样的输出，说明最小链路已经通了：

任务状态: completed 翻译结果: 你好 世界 欢迎 到 协议 协作

5.4 跑通之后的验证与扩展

跑通基础版的顺序是这样的：

1. 确认服务端日志里有收到tasks/send请求；
2. 确认客户端能正确解析任务状态和artifacts；
3. 在服务端加一个tasks/get接口，让客户端改用“先send再轮询get”的方式，模拟异步长任务；
4. 扩展Transport层，加一个SSE端点，把working状态实时推出去。

我自己做第一个A2A Demo时，从写代码到跑通大概花了一个多小时。大部分时间其实不是花在业务代码上，而是在确认“JSON-RPC字段有没有对齐”。所以真心建议你动手时先把两边的请求/响应报文打出来，一对一发消息先确认协议层没问题，再写业务逻辑。

6. 联调实录：几个让我反复排查的坑

协议的好处是规范统一，但联调的时候，坑往往藏在“规范没说死”的细节里。这里分享几个我实际遇到过的鬼问题，每一个都让我查了不少于半小时。

6.1 Agent Card里的URL写成了127.0.0.1

最经典的坑。本地起服务时，我在Agent Card的url字段写了http://127.0.0.1:8000。单机调试没问题，但一旦把客户端挪到另一台机器，或者跑在容器里，对方拿到这张卡片后怎么都连不上我的Agent。

这不是协议问题，是基础设施常识问题。但因为它太常见了，我还是要提。Agent Card里的地址应该是其他Agent“能访问到”的地址。在本机联调可以写localhost，跨机器联调务必写局域网IP或域名，容器环境就要用服务名或映射后的公网入口。

建议做法：把Agent Card里的URL做成环境变量，不同环境注入不同值。不要手改，太容易漏。

6.2 消息ID不唯一导致任务状态错乱

这是我在一个模拟项目X里最深刻的教训。当时写客户端代码，图省事把JSON-RPC的id写成了固定字符串，比如"message-001"。服务端倒也没报错，但任务状态查询返回的结果经常对不上。

查到最后才发现，服务端的框架用id作为请求关联的键，固定ID导致后到的响应覆盖了先到响应的结果。明明两个请求都对，返回的数据却张冠李戴。

解决很简单：客户端每个请求的id都生成唯一值，比如UUID。我用的是：

request_id = str(uuid.uuid4())

同时，taskId也要保证唯一。如果你同时向同一个Agent发多个任务，taskId重复会导致Agent端的任务状态互相覆盖。这个基本要靠调用方自己保证。

6.3 上下文过大与部分失败

A2A消息里的parts是可以携带大量文本的。我在一次做长文档翻译时，把整份文档塞进一条消息，结果请求体到了好几MB，服务端处理超时，整个任务失败。

后来我调整了策略：把大文档拆成多段，Agent之间按段传递任务；或者先用文件上传把文档放出来，消息里只带文件URL，让Agent自己去拉取。

经验是：文本尽量只放“指令和关键信息”，大块内容用文件引用代替。协议虽然允许你放很大的text，但不代表设计里期待你这么用。用file类型 + 临时URL的方案，既减少了消息体积，又天然支持了Agent从远端拉取文件的异步流程。

6.4 结果用text带过，下游Agent无法结构化消费

早期写Demo时，我的翻译Agent把结果放在一个text part里：“翻译完成，结果是……”。下游Agent如果只想拿最终译文，还得自己从那句自然语言里抠文本，非常蠢。

后来改成把结果放在artifacts里，用data类型或file类型统一表达，下游Agent直接解析task.artifacts[0].content就能拿到干净结果。这就是我在2.3节强调的边界问题，实践里极其关键。

我的建议是：在Agent开发规范里直接约定，凡是需要下游继续消费的产出，一律进artifacts；凡是人看的自然语言反馈，才进message parts。这样你每次联调都不用猜对方返回的东西该往哪取。

结尾

到目前为止，A2A协议的实践派思路已经比较完整了：从Agent Card建立身份和技能认知，到JSON-RPC消息让Agent之间能说话，再到任务状态机驱动多轮协作，最后加上鉴权边界保护整个网络。这套链路我前后折腾了近两周，才从“看得懂协议文档”变成“能跟Agent真的对话”。

我个人的体会是，A2A的最大价值不在于消息格式多么精致，而在于它给Agent协作提供了一个“可测试的沟通基线”。只要每个Agent都遵循同一套消息和状态约定，独立开发和部署的Agent就能组成临时的协作网络，而不用提前约定彼此的内部API。

如果你打算在真实项目里落地，我建议不要一上来就照搬整套规范。先只实现tasks/send和tasks/get，用轮询跑通一条最小业务链路；再逐步补上streaming、push通知和OAuth2。协议是工具，不是门面，真正让协作变可靠的是你在状态管理和消息设计里的每一次取舍。之后我会再写一篇关于Agent协作中数据流治理和错误恢复策略的内容，继续把这块做实。