8.3 跨语言 Agent：TS 实现 A2A 端与 Python 协作
8.1 的 A2A 服务端是 Python 写的，8.2 又讲清了协议分层。现在到了兑现协议价值的时候：真实团队里，写编排层的是 Python，而写前端服务、边缘函数、Node 后端的是 TypeScript。如果协作必须依赖同一门语言，那协议就没有意义。 协议的意义恰恰在于——只要双方说同一种线上语言，底层实现是 Python、TypeScript 还是 Go 都不重要。
本节把「协议抽象」兑现成一次真跑的跨语言协作：用 TypeScript（Node 26，只用可擦除语法、零 npm 依赖）实现一个 A2A 端，先让它被 8.1 的 Python 客户端调用，再让 TS 客户端反过来调用 8.1 的 Python 服务端，两个方向都真跑并贴出输出；顺带用
tsc --noEmit证明这段 TS 能过类型检查。
8.3.1 落点在「协议」而不是「绑定」
很多框架会提供「跨语言 SDK」，但那本质是语言绑定：换一门语言就得等一门官方 SDK，还得忍受版本不同步。A2A 走的是另一条路——只规定线格式。8.1 的服务端是 100 行 Python 标准库，本节的服务端是同样 100 行左右的 TS，两者互不认识对方的存在，却因为共享同一套 Message / Task / Agent Card 形状而天然能对话。这就是「协议 vs 绑定」的差别：绑定要求同源，协议只要求同形。
8.3.2 TS 端：Node 26 内置模块实现 A2A
Node 26 支持直接运行 .ts（类型擦除），所以下面这段代码不需要任何构建步骤、也不需要 npm 装包——只用内置的 node:http 与 node:crypto。注意它刻意只用可擦除语法：不用 enum、不用 namespace、不用构造函数参数属性，这样 node a2a_server.ts 才能零构建跑起来。
// A2A 最小服务端（TypeScript / Node 26，仅用内置 http，无 npm 依赖）
import { createServer } from "node:http";
import { createHash } from "node:crypto";
const PORT = 8742;
type Part = { kind: string; text?: string; data?: unknown };
type Message = { role: string; messageId: string; parts: Part[]; taskId?: string; contextId?: string };
type Skill = { id: string; name: string; description: string; tags: string[] };
type AgentCard = {
protocolVersion: string; name: string; description: string; url: string;
version: string; capabilities: Record<string, boolean>;
defaultInputModes: string[]; defaultOutputModes: string[]; skills: Skill[];
};
const AGENT_CARD: AgentCard = {
protocolVersion: "0.3.0",
name: "ResearchCopilot-TS",
description: "本地研究助理 Agent（TypeScript 端），提供确定性关键词抽取技能。",
url: `http://127.0.0.1:${PORT}/`,
version: "0.1.0",
capabilities: { streaming: false, pushNotifications: false },
defaultInputModes: ["text/plain"],
defaultOutputModes: ["application/json"],
skills: [
{
id: "keywords",
name: "关键词抽取",
description: "按词频确定性抽取关键词，返回 JSON Artifact。",
tags: ["text", "keywords"],
},
],
};
function runSkill(text: string): Record<string, unknown> {
const tokens = text.toLowerCase().match(/[a-z0-9]+/g) ?? [];
const freq = new Map<string, number>();
for (const t of tokens) freq.set(t, (freq.get(t) ?? 0) + 1);
const top = [...freq.entries()].sort((a, b) => b[1] - a[1] || a[0].localeCompare(b[0])).slice(0, 3);
const fingerprint = createHash("sha256").update(text).digest("hex").slice(0, 12);
return { tokens: tokens.length, keywords: top.map(([w, n]) => `${w}:${n}`), fingerprint };
}
function handleMessageSend(params: { message: Message }) {
const message = params.message;
const text = message.parts.filter((p) => p.kind === "text").map((p) => p.text ?? "").join("");
return {
id: message.taskId ?? createHash("md5").update(String(Date.now())).digest("hex").slice(0, 8),
contextId: message.contextId ?? "ctx-ts",
status: { state: "completed", timestamp: "2026-10-07T18:00:00+08:00" },
history: [message],
artifacts: [
{ artifactId: "a1", name: "keywords.json", parts: [{ kind: "data", data: runSkill(text) }] },
],
};
}
function sendJson(res: import("node:http").ServerResponse, code: number, payload: unknown): void {
const body = Buffer.from(JSON.stringify(payload), "utf-8");
res.writeHead(code, { "Content-Type": "application/json; charset=utf-8", "Content-Length": body.length });
res.end(body);
}
const server = createServer((req, res) => {
if (req.method === "GET" && req.url === "/.well-known/agent-card.json") {
sendJson(res, 200, AGENT_CARD);
return;
}
if (req.method === "POST") {
const chunks: Uint8Array[] = [];
req.on("data", (c: Uint8Array) => chunks.push(c));
req.on("end", () => {
const rpc = JSON.parse(Buffer.concat(chunks).toString("utf-8") || "{}");
if (rpc.method !== "message/send") {
sendJson(res, 200, { jsonrpc: "2.0", id: rpc.id, error: { code: -32601, message: "method not found" } });
return;
}
sendJson(res, 200, { jsonrpc: "2.0", id: rpc.id, result: handleMessageSend(rpc.params) });
});
return;
}
sendJson(res, 404, { error: "not found" });
});
server.listen(PORT, "127.0.0.1", () => {
console.log(`Agent Card: http://127.0.0.1:${PORT}/.well-known/agent-card.json`);
});
把它和 8.1 的 Python 端并排看，会发现协议层完全同构：同样两个路由、同样一个 run_skill、同样把结果包成 Task。语言换了，线格式一个字没变。
8.3.3 类型检查：erasableSyntaxOnly
上面这段代码要能过 tsc --noEmit，靠的是 tsconfig.json 里打开 erasableSyntaxOnly，保证它不掺入任何「需要代码生成」的语法（enum / namespace / 参数属性）：
{
"compilerOptions": {
"target": "esnext",
"module": "nodenext",
"moduleResolution": "nodenext",
"strict": true,
"noEmit": true,
"verbatimModuleSyntax": true,
"erasableSyntaxOnly": true,
"allowImportingTsExtensions": true,
"skipLibCheck": true
},
"include": ["**/*.ts"]
}
本机实测（typescript 7.0.2）：
$ npx tsc --noEmit
（无输出，退出码 0）
一个真实的踩坑：本机没有装 @types/node，而 tsc 需要它才能识别 node:http / node:crypto 的类型，直接编译会报 Cannot find module 'node:http'。解决办法是在工程里放一个极小的 ambient 声明文件（declare module "node:http" { ... }），只声明本例用到的几个符号。Node 的运行时并不需要它——类型擦除时这些声明会被整段丢弃，它纯粹是给 tsc 看的。这也侧面说明：TS 端跑 A2A 真的零依赖，连类型包都可以自己补。
8.3.4 真跑方向一：Python 客户端 → TS 服务端
启动 TS 服务端后，用 Python 客户端来调它——调用方是 Python，被调方是 Node/TS：
"""Python 客户端跨语言调用 TS A2A 端：取卡 -> 发任务 -> 读 Artifact。"""
import json
import urllib.request
BASE = "http://127.0.0.1:8742"
def get_json(url):
with urllib.request.urlopen(url, timeout=5) as r:
return json.loads(r.read().decode("utf-8"))
def post_json(url, payload):
req = urllib.request.Request(url, data=json.dumps(payload).encode("utf-8"),
headers={"Content-Type": "application/json"})
with urllib.request.urlopen(req, timeout=5) as r:
return json.loads(r.read().decode("utf-8"))
card = get_json(f"{BASE}/.well-known/agent-card.json")
print("跨语言发现 Agent:", card["name"], "| 运行时: Node/TS")
print("技能:", [s["id"] for s in card["skills"]])
rpc = {"jsonrpc": "2.0", "id": 7, "method": "message/send",
"params": {"message": {"role": "user", "messageId": "py-1",
"parts": [{"kind": "text",
"text": "Agent protocol agent protocol task"}]}}}
task = post_json(card["url"], rpc)["result"]
print("任务状态:", task["status"]["state"])
print("Artifact:", json.dumps(task["artifacts"][0]["parts"][0]["data"], ensure_ascii=False))
本机实测输出：
跨语言发现 Agent: ResearchCopilot-TS | 运行时: Node/TS
技能: ['keywords']
任务状态: completed
Artifact: {"tokens": 5, "keywords": ["agent:2", "protocol:2", "task:1"], "fingerprint": "affd4c9bc6f5"}
Python 客户端拿到的 Task 形状，与它调自家 Python 服务端时一模一样——这正是协议抽象在起作用。
8.3.5 真跑方向二：TS 客户端 → Python 服务端
反过来也一样。用 TypeScript 写一个客户端，去调 8.1 那个 Python 服务端（Node 26 内置 fetch，同样零依赖）：
// A2A 客户端（TypeScript）：发现 Python 端 Agent Card，并委派一个任务。
const BASE = "http://127.0.0.1:8741";
type AgentCard = { name: string; url: string; protocolVersion: string; skills: { id: string }[] };
type Task = { status: { state: string }; artifacts: { parts: { data: unknown }[] }[] };
const card: AgentCard = await (await fetch(`${BASE}/.well-known/agent-card.json`)).json();
console.log("发现 Agent:", card.name, "| 协议", card.protocolVersion);
const rpc = {
jsonrpc: "2.0",
id: 42,
method: "message/send",
params: {
message: {
role: "user",
messageId: "ts-1",
parts: [{ kind: "text", text: "TS 客户端委派给 Python Agent" }],
},
},
};
const resp = await fetch(card.url, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify(rpc),
});
const { result }: { result: Task } = await resp.json();
console.log("任务状态:", result.status.state);
console.log("Artifact:", JSON.stringify(result.artifacts[0].parts[0].data));
本机实测输出（TS 进程调 Python 服务端）：
发现 Agent: ResearchCopilot-Py | 协议 0.3.0
任务状态: completed
Artifact: {"chars":22,"words":4,"fingerprint":"c44255dae99f","summary":"共 22 字符、4 个词，指纹 c44255dae99f"}
到这里，两个方向都跑通了：Python 客户端调 TS 服务端、TS 客户端调 Python 服务端。同一套线格式，四种进程组合里没有一种是需要额外适配的。
8.3.6 跨语言协作的四个坑
跨语言不是「写对了就能通」，下面四点在本例里都被真实地碰到过：
| 坑 | 具体表现 | 对策 |
|---|---|---|
| 消息形状必须逐字一致 | parts 的 kind、role 取值一处不同就解析失败 | 把线格式写成共享的 schema，双方照抄 |
JSON-RPC 的 id 必须回填 | 并发请求时靠 id 对号入座，回错就串响应 | 服务端原样回传 req.id（本例两端都做） |
| 字段命名风格 | Python 习惯 snake_case，TS 习惯 camelCase | 协议层统一成一种（本例统一 camelCase） |
| 数值与编码 | 中文/emoji 的字节数、浮点精度两端算法可能不同 | 显式 utf-8；统计类字段用整数；必要时比对指纹 |
第一条尤其致命：A2A 的 Message.parts[].kind 是 text / file / data 三选一，Python 端若漏写 kind，TS 端 filter((p) => p.kind === "text") 就会把整条消息过滤掉，表现为「任务成功但 Artifact 是空的」——一个没有任何报错的静默失败。
8.3.7 这套协议替你做了什么、你又让渡了什么
A2A 替你做的：让「跨语言协作」退化成「跨语言抄一份线格式」。本例的 TS 端没有引入任何 A2A 官方库，只用内置 http 就实现了对端——因为协议只约束消息长什么样，不约束用什么实现。相比「等一门官方 SDK」，这是巨大的自由度。
你让渡的：自由度是有代价的。没有 SDK 意味着没有类型校验、没有重试、没有鉴权封装——这些在官方 SDK 里白送的东西，你都得自己写，而且要在每一门语言里各写一遍。本例能跑通，是因为协议被简化到了最小子集；真上生产，你要在 Python 与 TS 两边各自实现超时、重试、错误码映射、鉴权，任何一边漏了，跨语言协作就会在边界处炸。所以务实的选择是：协议保证互通，SDK 保证效率——先用协议把边界定死，再在有官方 SDK 的语言上用它省事，在没有的语言上照协议手写。
下一章 9.1 问题设定：轨迹、奖励与信用分配 会从「多个 Agent 怎么协作」转向「一个 Agent 怎么被训练得更好」。想回看 A2A 与 MCP 的边界，见 8.2 与 MCP 分工 。
阅读导航：上一节：8.2 与 MCP 分工：工具协议 vs 协作协议 · 下一节：9.1 问题设定：轨迹、奖励与信用分配 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。