用 ATP 搭一条服务端中转的多智能体通信链路

多智能体协作卡住的地方，往往不是模型能力，而是"消息怎么送到"。两个 Agent 各自跑在容器、内网或本地不同端口上，彼此没有可达地址，谁也不知道对方还在不在线。这篇用 ATP（Agent Transfer Protocol，智能体传输协议）搭一条最小可跑的链路：一个中转服务器负责注册与路由，两个 Agent 通过它交换消息，并为每条消息带回执和超时判定。

不同实现或版本的 ATP 在字段命名、状态码上会有差异，本文使用的是一套通用子集，对接正式实现时以官方规范当前版本为准。

这篇能做出什么

跟着做完，你会得到三个文件、约两百行 Python，以及一条能反复验证的链路：

• worker Agent 启动时把自己的回调地址注册到中转服务器，之后能收到别人发给它的任务；
• planner Agent 只负责发，把任务投给中转服务器，拿到"对方已收到并处理"的回执；
• 中转服务器记录每个 Agent 的在线状态，投递失败先进队列，按间隔重投，超过 TTL 判定超时；
• 你能亲手制造一次"接收方离线"，观察消息从 queued 变成 timeout，再把 worker 拉起来重跑一遍，看到恢复。

实测时的终端输出会长这样：

```text

注册: {'ok': True, 'agent_id': 'planner', 'lease_seconds': 300, 'queued': 0}

发送任务…

第 1 次投递 → 200 delivered

最终结果: {'status': 'delivered', 'message_id': '...', 'receipt': {'status': 'accepted', 'handled_by': 'worker', 'result': {'text': 'yaler aiv tnegam-itlum'}}}

```

整条链路只用到 HTTP 和 JSON，不需要公网、不需要任何密钥。

前置条件清单

动手前确认这几件事：

1. 本机装了 Python 3.9 及以上版本（具体版本以官方文档当前版本为准），能执行 python --version。

2. 会用终端，能同时开三个终端窗口（或三个 tmux 面板）。

3. 知道 HTTP 请求、JSON、进程端口这些基本概念。

4. 端口 8001、8002、8010 没被别的程序占用；被占了就换，代码里对应的常量一起改。

5. 目录结构如下，三个文件放在同一层：

```text

atp-demo/

├── relay_server.py # 中转服务器

├── worker_agent.py # 接收侧 Agent

└── planner_agent.py # 发送侧 Agent

```

装依赖（不指定版本，让 pip 取当前稳定版）：

```bash

python -m venv .venv

source .venv/bin/activate # Windows 用 .venv\Scripts\activate

pip install fastapi uvicorn httpx

```

分步骤

步骤 1：先把链路画清楚，再写代码

一条消息在系统里走三段：

1. 注册：Agent 启动时向中转服务器上报 agent_id 和 callback_url，同时给一个租约时长。租约到期没续，服务器就当它离线。

2. 路由：发送方把消息投给中转服务器，服务器查注册表，找到接收方的回调地址，用 HTTP POST 推过去。

3. 回执：接收方处理完，在 HTTP 响应体里返回一条回执。服务器保存回执，发送方按 message_id 查询。

这里有个关键区别，务必先分清：

• 传输回执：服务器告诉你"这条消息我已经接住了"或者"已经送到对方机器上了"。
• 业务回执：接收方告诉你"这个任务我做完了，结果是什么"。

本文把两者压进一次交互里方便演示，生产环境建议拆开，后面"下一步建议"会讲怎么改。

步骤 2：约定消息信封

所有消息都套一层统一的信封，业务内容放 payload。这样中转服务器不需要理解业务，只看信封就能路由。

```json

{

"atp_version": "1.0",

"message_id": "550e8400-e29b-41d4-a716-446655440000",

"sender": "planner",

"receiver": "worker",

"type": "task.request",

"timestamp": 1730000000.0,

"ttl_ms": 5000,

"correlation_id": null,

"payload": { "text": "hello" }

}

```

字段职责：

步骤 3：写中转服务器

relay_server.py，负责注册、路由、重投、回执查询四件事。为了能直接跑，注册表和队列都放内存，重启即清空。

```python

"""

relay_server.py —— ATP 中转服务器（内存版，仅用于本地跑通链路）

运行：uvicorn relay_server:app --port 8001

"""

import asyncio

import time

import uuid

from collections import defaultdict, deque

from typing import Any, Dict, List, Optional

import httpx

from fastapi import FastAPI, HTTPException

from pydantic import BaseModel, Field

app = FastAPI(title="ATP Relay Server")

AGENTS: Dict[str, Dict[str, Any]] = {} # agent_id -> 注册信息

QUEUE: Dict[str, deque] = defaultdict(deque) # agent_id -> 待重投消息

RECEIPTS: Dict[str, Dict[str, Any]] = {} # message_id -> 回执状态

SEEN: Dict[str, float] = {} # message_id -> 首次收到时间（幂等）

RETRY_INTERVAL_SECONDS = 1.0

DELIVER_TIMEOUT_SECONDS = 10.0

class RegisterBody(BaseModel):

agent_id: str

callback_url: str

capabilities: List[str] = Field(default_factory=list)

lease_seconds: int = 300

class Envelope(BaseModel):

atp_version: str = "1.0"

message_id: str = Field(default_factory=lambda: str(uuid.uuid4()))

sender: str

receiver: str

type: str

timestamp: float = Field(default_factory=lambda: time.time())

ttl_ms: int = 5000

correlation_id: Optional[str] = None

payload: Dict[str, Any] = Field(default_factory=dict)

def expired(self) -> bool:

return (time.time() - self.timestamp) * 1000 > self.ttl_ms

def live_agent(agent_id: str) -> Optional[Dict[str, Any]]:

rec = AGENTS.get(agent_id)

if rec and rec["expires_at"] > time.time():

return rec

return None

async def deliver(callback_url: str, env: Envelope):

"""推一次。返回 (是否成功, 回执或错误信息)。"""

try:

async with httpx.AsyncClient(timeout=DELIVER_TIMEOUT_SECONDS) as client:

resp = await client.post(callback_url, json=env.model_dump())

if resp.status_code == 200:

return True, resp.json()

return False, {"error": f"http_{resp.status_code}"}

except Exception as exc: # 拒绝连接、超时、DNS 失败都走这里

return False, {"error": type(exc).__name__}

@app.post("/atp/register")

async def register(body: RegisterBody):

AGENTS[body.agent_id] = {

"callback_url": body.callback_url,

"capabilities": body.capabilities,

"expires_at": time.time() + body.lease_seconds,

}

return {"ok": True, "agent_id": body.agent_id,

"lease_seconds": body.lease_seconds,

"queued": len(QUEUE.get(body.agent_id, []))}

@app.get("/atp/agents")

async def list_agents():

now = time.time()

return {aid: {**rec, "alive": rec["expires_at"] > now}

for aid, rec in AGENTS.items()}

@app.post("/atp/send")

async def send(env: Envelope):

1) 幂等：同一个 message_id 只路由一次

if env.message_id in SEEN:

cached = RECEIPTS.get(env.message_id, {"status": "queued", "receipt": None})

return {"status": cached["status"], "message_id": env.message_id,

"receipt": cached.get("receipt"), "duplicate": True}

SEEN[env.message_id] = time.time()

2) 进来就过期的，直接判死

if env.expired():

RECEIPTS[env.message_id] = {"status": "expired", "receipt": None}

return {"status": "expired", "message_id": env.message_id, "receipt": None}

3) 找接收方

target = live_agent(env.receiver)

if target is None:

RECEIPTS[env.message_id] = {"status": "unreachable", "receipt": None}

return {"status": "unreachable", "message_id": env.message_id, "receipt": None}

4) 先推一次，推不到就入队等重投

ok, receipt = await deliver(target["callback_url"], env)

if ok:

RECEIPTS[env.message_id] = {"status": "delivered", "receipt": receipt}

return {"status": "delivered", "message_id": env.message_id, "receipt": receipt}

QUEUE[env.receiver].append(env.model_dump())

RECEIPTS[env.message_id] = {"status": "queued", "receipt": None,

"last_error": receipt}

return {"status": "queued", "message_id": env.message_id, "receipt": None}

@app.get("/atp/receipts/{message_id}")

async def get_receipt(message_id: str):

rec = RECEIPTS.get(message_id)

if rec is None:

raise HTTPException(status_code=404, detail="unknown message_id")

return {"message_id": message_id, **rec}

@app.get("/atp/pending/{agent_id}")

async def pull_pending(agent_id: str):

"""拉模式：连不上回调地址的 Agent 也可以主动来取。"""

q = QUEUE.get(agent_id)

if not q:

return {"agent_id": agent_id, "messages": []}

items = []

while q:

items.append(q.popleft())

return {"agent_id": agent_id, "messages": items}

async def retry_loop():

"""每秒扫一遍队列：没过期的重投，过了期的判死信。"""

while True:

await asyncio.sleep(RETRY_INTERVAL_SECONDS)

for agent_id in list(QUEUE.keys()):

target = live_agent(agent_id)

q = QUEUE[agent_id]

keep = deque()

while q:

raw = q.popleft()

env = Envelope(**raw)

if env.expired():

RECEIPTS[env.message_id] = {"status": "timeout", "receipt": None}

continue

if target is None:

keep.append(raw)

continue

ok, receipt = await deliver(target["callback_url"], env)

if ok:

RECEIPTS[env.message_id] = {"status": "delivered", "receipt": receipt}

else:

keep.append(raw)

QUEUE[agent_id] = keep

@app.on_event("startup")

async def _startup():

asyncio.create_task(retry_loop())

```

如果你的 pydantic 还是 v1，把两处 model_dump() 换成 dict()。

步骤 4：写接收侧 Agent

worker_agent.py，两个要点：启动时自动注册（带重试，因为中转服务器可能还没起来），处理消息时按 message_id 幂等。

```python

"""

worker_agent.py —— 接收侧 Agent

运行：uvicorn worker_agent:app --port 8002

"""

import asyncio

import os

import time

from typing import Any, Dict, Set

import httpx

from fastapi import FastAPI

RELAY = os.getenv("ATP_RELAY", "http://127.0.0.1:8001")

AGENT_ID = os.getenv("ATP_AGENT_ID", "worker")

CALLBACK = os.getenv("ATP_CALLBACK", "http://127.0.0.1:8002/atp/inbox")

app = FastAPI(title=f"ATP Agent {AGENT_ID}")

SEEN: Set[str] = set()

@app.post("/atp/inbox")

async def inbox(env: Dict[str, Any]) -> Dict[str, Any]:

mid = env["message_id"]

幂等：重投可能把同一条消息送多次

if mid in SEEN:

return {"status": "duplicate", "message_id": mid, "code": 200,

"detail": "已处理过，直接返回回执"}

if env["type"] != "task.request":

return {"status": "rejected", "message_id": mid, "code": 415,

"detail": f"不认识的 type: {env['type']}"}

text = env.get("payload", {}).get("text", "")

if not text:

return {"status": "rejected", "message_id": mid, "code": 400,

"detail": "payload.text 为空"}

SEEN.add(mid)

await asyncio.sleep(0.3) # 假装在干活

return {"status": "accepted", "message_id": mid, "code": 200, "detail": "ok",

"handled_by": AGENT_ID, "handled_at": time.time(),

"result": {"text": text[::-1]}}

@app.on_event("startup")

async def register_self():

async with httpx.AsyncClient(timeout=5.0) as client:

for _ in range(10):

try:

r = await client.post(f"{RELAY}/atp/register", json={

"agent_id": AGENT_ID,

"callback_url": CALLBACK,

"capabilities": ["text.reverse"],

"lease_seconds": 300,

})

if r.status_code == 200:

print(f"[{AGENT_ID}] 注册成功: {r.json()}")

return

except Exception as exc:

print(f"[{AGENT_ID}] 注册失败({type(exc).__name__})，1 秒后重试")

await asyncio.sleep(1.0)

print(f"[{AGENT_ID}] 多次注册失败，请检查中转服务器是否已启动: {RELAY}")

```

步骤 5：写发送侧 Agent

planner_agent.py 只发不收，不需要起 HTTP 服务，直接当脚本跑。它要处理三件事：投递、轮询回执、失败重试。

```python

"""

planner_agent.py —— 发送侧 Agent（只发不收）

运行：python planner_agent.py

"""

import asyncio

import time

import uuid

import httpx

RELAY = "http://127.0.0.1:8001"

ME = "planner"

WORKER = "worker"

MY_CALLBACK = "http://127.0.0.1:8010/atp/inbox" # 本 Demo 不监听，占位

TERMINAL = {"timeout", "unreachable", "expired"}

async def register(client):

r = await client.post(f"{RELAY}/atp/register", json={

"agent_id": ME,

"callback_url": MY_CALLBACK,

"capabilities": ["plan"],

"lease_seconds": 300,

})

r.raise_for_status()

print("注册:", r.json())

async def wait_receipt(client, message_id, timeout_s):

deadline = time.time() + timeout_s

while time.time() < deadline:

r = await client.get(f"{RELAY}/atp/receipts/{message_id}")

if r.status_code == 200:

data = r.json()

if data["status"] in ("delivered", "timeout", "unreachable", "expired"):

return data

await asyncio.sleep(0.2)

return {"status": "client_timeout", "message_id": message_id, "receipt": None}

async def send_task(client, text, ttl_ms=5000, max_attempts=3):

message_id = str(uuid.uuid4())

for attempt in range(1, max_attempts + 1):

env = {

"atp_version": "1.0",

"message_id": message_id,

"sender": ME,

"receiver": WORKER,

"type": "task.request",

"timestamp": time.time(),

"ttl_ms": ttl_ms,

"payload": {"text": text},

}

resp = await client.post(f"{RELAY}/atp/send", json=env)

print(f" 第 {attempt} 次投递 → {resp.status_code} {resp.json().get('status')}")

result = await wait_receipt(client, message_id, timeout_s=ttl_ms / 1000 + 2)

if result["status"] == "delivered":

return result

print(f" 没拿到回执：{result['status']}")

if result["status"] in TERMINAL:

服务端已判终态，同一个 id 不会再被路由，换新的重来

message_id = str(uuid.uuid4())

await asyncio.sleep(min(0.5 * 2 ** (attempt - 1), 4.0))

return {"status": "failed", "message_id": message_id}

async def main():

async with httpx.AsyncClient(timeout=15.0) as client:

await register(client)

print("发送任务…")

result = await send_task(client, "multi agent via relay")

print("最终结果:", result)

if __name__ == "__main__":

asyncio.run(main())

```

步骤 6：跑通第一遍

开三个终端，都先 source .venv/bin/activate。

```bash

终端 1：中转服务器

uvicorn relay_server:app --port 8001

```

```bash

终端 2：接收侧

uvicorn worker_agent:app --port 8002

```

```bash

终端 3：发送侧

python planner_agent.py

```

发送侧应该打印 delivered，并在 receipt.result.text 里看到被倒序的字符串。再查一下注册表：

```bash

curl -s http://127.0.0.1:8001/atp/agents | python -m json.tool

```

能看到 planner 和 worker 两条记录，alive 都是 true。

步骤 7：故意制造一次超时

这一步是整篇的重点——没验证过失败路径的通信链路不算跑通。

在终端 2 按 Ctrl+C 停掉 worker，然后再跑一次 python planner_agent.py。观察输出：

```text

第 1 次投递 → 200 queued

没拿到回执：client_timeout

第 2 次投递 → 200 queued

没拿到回执：timeout

第 3 次投递 → 200 timeout

没拿到回执：timeout

最终结果: {'status': 'failed', ...}

```

queued 说明服务器接住了但没送到；timeout 说明重投到 TTL 到期还没成功，消息进死信。注意第 2 次之后换了一个新的 message_id，因为旧的那条已经被服务端判终态，复用 id 只会拿到缓存的旧回执。

再把 worker 拉起来，重新跑一次，链路恢复。这就验证了三件事：失败时消息不会凭空消失、TTL 到期会有确定结论、接收方恢复后能自动续上。

步骤 8：并发发多轮任务

把 main() 换成并发版本，模拟一个 Planner 同时调度多个任务：

```python

async def main():

async with httpx.AsyncClient(timeout=15.0) as client:

await register(client)

results = await asyncio.gather(

send_task(client, "alpha"),

send_task(client, "beta"),

send_task(client, "gamma"),

)

for r in results:

if r["status"] == "delivered":

print(r["receipt"]["result"])

```

三个任务各自拿独立的 message_id 并发走同一条中转链路，互不干扰。真实的多智能体协作里，这一步就是"一个 Planner 并行派活给多个 Worker"的雏形。

常见坑与排错

1. 一开始能发，几分钟后突然 unreachable

注册有租约（本文是 300 秒），到期没续就视为离线。用 GET /atp/agents 看 alive 字段确认。生产上要么加心跳接口，要么把租约拉长并在 Agent 侧定时续约。

2. 中途改了代码、重启了 uvicorn --reload，worker 就收不到消息了

--reload 会重启整个进程，内存里的注册表没了，服务器自然找不到人。worker 需要重新注册。开发时建议中转服务器别开 --reload，或者干脆把注册动作做成定时续约而不是只在上启动时做一次。

3. 把回调地址写成 http://127.0.0.1:8002 但在容器里跑

容器内的 127.0.0.1 指的是容器自己。容器之间要用服务名或容器网络里的地址。本地裸跑没问题，一进 Docker 就是头号故障源。

4. 消息被处理了两次

服务端的 SEEN 只减少重复路由，不能替代接收侧幂等：网络超时导致服务器以为没送到、实际对方已经处理，重投就会重复。接收侧必须按 message_id 去重，本文用 SEEN 集合演示，生产上换成带过期时间的 Redis key。

5. 客户端等不到 timeout 回执

客户端等待时间必须大于 ttl_ms，否则服务端还没来得及判死信，你就先放弃了。本文里 wait_receipt 用的是 ttl_ms/1000 + 2，就是这个道理。反过来，服务端投递的 HTTP 超时也要小于 TTL，不然重投根本没机会发生。

6. 用时间戳判过期，跨机部署时判不准

timestamp 由发送方生成，两台机器时钟差几秒，TTL 就偏了。稳妥做法是服务端收到时记一个 received_at，用自己的时钟算过期。

7. 内存版重启全丢

注册表、队列、回执都在进程内存里，重启即清空。这只是为了把链路跑通，别拿去生产用。

下一步建议

链路跑通之后，按需要往几个方向加厚：

• 换存储：注册表放 Redis 的 hash，待投队列用 Stream 或 List，回执单独存一份带 TTL 的 key，这样中转服务器可以多实例部署。
• 加鉴权：注册时校验凭证，消息带发送方签名，防止任意进程冒充别人投递。具体签名方案以 ATP 官方规范当前版本为准。
• 拆分回执与结果：回执只表示"收到/拒绝/处理中"，真正的结果用另一条 task.result 消息发回来，用 correlation_id 串成一次会话。这样长任务才不会把回执通道堵死。
• 加能力发现：注册时的 capabilities 目前只是存着，可以加 GET /atp/agents?capability=text.reverse，让 Planner 按能力选 Worker，而不是写死 id。
• 换传输方式：回调对 NAT 和容器网络不友好，可以换成 WebSocket 或 SSE 长连接，Agent 主动连上来，服务器直接推。
• 加可观测：给每条消息打一个 trace id，记录投递耗时、重投次数、超时率。多智能体系统出问题时，第一件事就是看这条链路的水位和延迟。

先把本文这条两三个 Agent 的最小链路跑顺、把失败路径验穿，再往上加东西，比一上来就搭大框架省事得多。