简介智能体通信协议正成为AI从单点工具走向协作网络的关键基础设施。这份PDF系统梳理了模型上下文协议MCP、智能体到智能体协议A2A、智能体网络协议ANP三大协议的发展背景与设计理念MCP以模型为中心实现大型语言模型应用与外部数据源/工具的无缝集成A2A面向企业内部复杂协作采用P2P架构ANP则以智能体为中心基于W3C DID和语义网技术解决身份、描述、发现等核心问题目标成为智能体互联网时代的HTTP。内容还给出了协议对比表格、ANP分层架构身份与加密通信层、元协议层、应用协议层、MCP与ANP区别例如“模型USB-C”与“智能体email”的比喻并结合开源Manus接入ANP等案例帮助AI研究人员、开发者及企业管理者在理解概念的基础上选择合适协议同时为探讨AI原生数据网络、消费互联网与产业互联网融合等前沿方向提供参考。全包仅1个PDF文件大小8.14MB已有583人学习下载适合作为了解智能体互联网络发展趋势的入门参考。1. 智能体互联网络靠什么串起来MCP、A2A、ANP 的分工与关系如果你已经习惯了把 MCP Server 接到 Cursor、Claude Code 或 Trae 里让 AI 直接调用本地工具那你其实已经在搭智能体互联网络的第一块砖了。但真要让多个 Agent 互相派活、共享工具、协商结果单靠 MCP 是不够的MCP 管的是“模型到工具的桥接”A2A 管的是“Agent 到 Agent 的协作”ANP 管的是“每个 Agent 在网络里的名字与地址”。这篇文章会把三者分层讲清楚从最小 MCP Server 写到 A2A 任务流转最后给你一份联调阶段的排错清单和验证技巧适合正在做多 Agent 编排、或者准备把单机工具接入到企业级 Agent 平台的开发者。2. 协议分层与选型MCP 管工具A2A 管协作ANP 管寻址三个协议经常被放在一起讨论但它们并不是同一个赛道上的竞品。MCP 解决的是“Agent 的手够不够长”A2A 解决的是“两个 Agent 怎么对话”ANP 解决的是“对方叫什么、在哪里”。你可以把这三者看成三个独立的分层工具能力层、协作消息层、命名寻址层。这一步没看清楚后面做架构设计时很容易把协议职责混在一起导致一个服务里既写 MCP 又写 A2A最后两边都不伦不类。2.1 MCP模型到工具的标准桥接生态最成熟MCPModel Context Protocol是当前落地最广的智能体工具协议。它采用 client-server 结构Agent 作为 MCP Client通过 JSON-RPC 调用 MCP Server 暴露的 tools、resources 和 prompts。Cursor、Claude Code、Codex、Trae 这些常见客户端都内置了 MCP Client所以你的工具只要包成一个 MCP Server就能同时被多个智能体使用不需要为每个平台单独写适配层。一个最小可跑的 MCP Server 用官方 Python SDK 写起来很短from mcp.server.fastmcp import FastMCP mcp FastMCP(ops-agent) mcp.tool() def get_city_temperature(city: str, unit: str celsius) - str: 查询指定城市的当前温度。unit 可选 celsius 或 fahrenheit。 # 真实项目里这里会去接气象服务或者内部监控系统 return f{city}: 22 {unit} if __name__ __main__: mcp.run(transportstdio)这段代码里有三个参数值得注意。第一个是FastMCP(ops-agent)里的字符串它是这个 Server 对外展示的名字会出现在客户端的 MCP 列表里。第二个是工具函数get_city_temperature的 docstring这行文本会被当作工具描述发给大模型描述写得越具体模型就越知道在什么场景下调用它。第三个是mcp.run(transportstdio)它表示通过标准输入输出传输适合本地命令行启动如果要多台机器共享这个工具可以改成transportstreamable_http让 Agent 通过 URL 访问。需要特别注意的是stdio 模式下 Server 的标准输出被协议占用任何print()都会污染 MCP 协议字节流。要打日志请用logging模块或者直接写 stderr。这一点我见过太多人踩后面避坑章节会专门展开。2.2 A2AAgent 到 Agent 的对等协作协议A2AAgent-to-Agent解决的是“Agent 之间怎么互相派活”。它不使用 MCP 那套 tools/call 模型而是让每个 Agent 对外暴露一张 Agent Card描述自己能做什么、接受什么输入、回调地址是什么。调用方通过 JSON-RPC 2.0 发送消息收到任务的一方返回一个 taskId然后双方基于这个任务状态继续协作。下面是 A2A 消息的一个最小 JSON 信封示例{ jsonrpc: 2.0, id: 1, method: message/send, params: { agentCardId: agent-b-001, message: { role: agent, content: { type: text, text: 请把这份巡检报告总结成 5 条风险要点并返回给调度 Agent } } } }注意这个结构跟 MCP 的tools/call完全不同MCP 是客户端请求一个明确命名的工具A2A 是向另一个 Agent 发送一条语义化消息由对方自己决定怎么处理。A2A 的任务状态机一般从 accepted 走到 working再到 completed 或 failed调用方通过tasks/get轮询进度也可以让被调用方通过回调 URL 主动推送结果。A2A 的成熟度目前不如 MCP但它是把单 Agent 变成多 Agent 协作的必要环节。你在做协议选型时只要出现“Agent 要主动把任务转交给另一个 Agent”的需求就应该引入 A2A而不是靠 MCP 去互相调用对方的内部工具。2.3 ANP名字到地址的解析层ANPAgent Name Protocol是最容易被忽视的一层。它的作用类似 DNS给每个 Agent 一个全网唯一的人类可读名字然后把它解析到当前实际可用的端点地址。为什么要单独做这一层因为 Agent 的部署地址会变本地开发时是localhost:8010上了 Kubernetes 后变成 Service 域名多云部署时可能要切到另一套网关。如果所有调用方都写死 IP 和端口每次地址变更都要改代码这个代价在 Agent 数量起来之后会非常可怕。ANP 的最小解析配置长这样{ agentName: ops-agent.main, endpoint: https://agent.internal.example.com/a2a, ttl: 60, signature: base64..., capabilities: [task/send, task/get] }调用方拿到ops-agent.main这个名字后先查 ANP 目录得到上面的记录再取出endpoint字段去发起 A2A 调用。ttl控制解析结果的缓存时长signature用来校验这条记录是否由该 Agent 的所有者签发防止别人伪造你的 Agent 地址。MCP 里没有这一层因为 MCP Client 是主动去连已知地址的 Server而 A2A 场景下调度方往往只知道“我要找订单处理 Agent”并不知道它当前部署在哪这时候 ANP 就补上了这层寻址能力。2.4 选型决策先 MCP 还是直接上 A2A/ANP在做架构设计时我一般按下面这张表来判断维度MCPA2AANP解决的核心问题模型如何调用工具Agent 之间如何协作Agent 名字如何寻址消息形态JSON-RPC tools/callJSON-RPC message/send解析记录查询典型使用者MCP ClientIDE、AgentAgent 调度方、被调用 Agent所有需要知道“对方在哪”的节点成熟度高生态最广中等规范仍在演进较新适合先做内部标准化上手成本低本地即可跑通中需要设计任务状态机中需要维护名字目录实际项目的常见做法是先只接 MCP把工具能力打通当第二个 Agent 出现并且需要第一个 Agent 帮它做事时再加 A2A当 Agent 数量超过十个、部署地址开始频繁变动时再引入 ANP。一上来就三件套全上只会把问题复杂化。3. 搭建最小智能体互联网络从 MCP Server 到 A2A 协作全流程这一章的目标是让你在本地完整跑通一条链路Agent 通过 MCP 调用工具再把结果通过 A2A 发给另一个 Agent最后用 ANP 做名字解析。你可以把这个最小网络当成后续扩展的骨架每一步都有明确的代码和参数。3.1 写一个真正能跑的最小 MCP Server上一章的示例是单工具版本实际联调时更常用的是多工具加日志输出。下面这段代码除了注册工具还把日志单独打到文件避免污染 stdioimport logging from logging.handlers import RotatingFileHandler from mcp.server.fastmcp import FastMCP # 日志走文件不要走 stdout否则 stdio 传输会被污染 handler RotatingFileHandler(mcp_ops.log, maxBytes5_000_000, backupCount2) logging.basicConfig( levellogging.INFO, format%(asctime)s %(levelname)s %(name)s %(message)s, handlers[handler] ) logger logging.getLogger(mcp-ops) mcp FastMCP(ops-agent) mcp.tool() def read_log_file(file_path: str, lines: int 20) - list[str]: 读取指定日志文件的最后 N 行用于排查线上问题。 logger.info(read_log_file called, path%s lines%d, file_path, lines) with open(file_path, r, encodingutf-8) as f: return f.readlines()[-lines:] mcp.tool() def restart_service(service_name: str) - str: 重启指定服务。仅限白名单服务名防止误操作。 allowed {web, worker, scheduler} if service_name not in allowed: return denied: service not in whitelist # 这里接 systemd 或 k8s 的重启接口 logger.warning(restart_service called, service%s, service_name) return f{service_name} restart accepted if __name__ __main__: mcp.run(transportstdio)这段代码比第一版多了两个工程化细节。一是RotatingFileHandler日志文件超过 5MB 自动滚动备份避免单文件无限增长二是restart_service里加了白名单校验这是工具层权限控制的常见做法——MCP 只负责把工具暴露出去谁有权调用、哪些参数合法应该在工具实现内自己把关。启动时先跑一次mcp dev ops_server.py这个命令会拉起一个本地调试面板让你直接查看工具列表和调用参数。确认没问题后再用python ops_server.py投到目标环境。mcp dev是官方 SDK 自带的调试入口能省掉很多“Client 看不到工具”的排查时间。3.2 在 Client 端注册 MCPtransport、超时与日志参数MCP Client 的配置文件在不同工具里字段大同小异。下面是一份通用 JSON 配置可以套到大多数支持 MCP 的客户端里{ mcpServers: { ops-agent: { command: python, args: [ops_server.py], env: { LOG_LEVEL: DEBUG }, timeout: 60 } } }command和args用于 stdio 模式Shell 会执行python ops_server.py并把它的 stdin/stdout 作为协议通道。env是给 Server 进程注入的环境变量适合透传 API Key 或控制日志级别但要注意不要把密钥写进会被提交到 Git 的配置里。timeout是调用工具的默认超时单位是秒如果工具本身执行很慢这个值要调大否则 Client 会先超时断开。如果你用的是远程 Server配置就不需要command改成一段 URL{ mcpServers: { ops-agent: { url: http://127.0.0.1:9000/mcp, timeout: 30 } } }这里的url指向 MCP Server 暴露的 streamable HTTP 端点。本地联调用 stdio 最省事因为它不需要准备端口和鉴权但一旦要多台机器共享一个 Server立刻要切到 HTTP 模式同时加上鉴权头和来源白名单。3.3 用 A2A 把两个 Agent 串起来任务发牌与结果回传MCP 打通之后下一步是让两个 Agent 能对话。下面是一个最小 A2A 服务端实现用 FastAPI 接收message/send请求立即返回 taskId再在后台执行任务# agent_b_a2a.py from fastapi import FastAPI, Request import uuid, asyncio app FastAPI() tasks {} async def run_task(task_id: str, payload: dict): # 模拟一个耗时的处理流程真实项目中这里会调用 LLM 或工具链 await asyncio.sleep(5) tasks[task_id] { taskId: task_id, status: {state: completed, message: 风险要点汇总完成} } app.post(/) async def a2a_endpoint(req: Request): body await req.json() if body.get(method) message/send: task_id str(uuid.uuid4()) tasks[task_id] {status: {state: working}} asyncio.create_task(run_task(task_id, body)) return { jsonrpc: 2.0, id: body.get(id), result: { taskId: task_id, status: {state: working} } } return {jsonrpc: 2.0, id: body.get(id), error: {code: -32601, message: method not found}}注意这里的任务状态机。调用方发出message/send后服务端立即返回working同时后台异步执行真正的业务逻辑。调用方随后可以请求tasks/get查询任务状态。之所以设计成异步返回而不是同步等到结果算完是因为 Agent 协作的任务往往要跑几十秒甚至几分钟同步等待会让调度方长时间阻塞。用 curl 做一次完整调用curl -s -X POST http://127.0.0.1:8010/ \ -H Content-Type: application/json \ -d { jsonrpc: 2.0, id: 1, method: message/send, params: { message: { role: agent, content: { type: text, text: 汇总巡检报告风险点 } } } }如果返回的result.status.state是working说明 A2A 链路已经通了。接下来调用方需要按间隔轮询tasks/get或者等服务端调用你在 Agent Card 里声明的回调地址。前者实现简单后者实时性好但必须做签名校验否则任何人都能伪造回调。3.4 ANP 解析怎么接进 A2A 路由当你有多个 Agent 地址要维护时直接在代码里写死 A2A 端点就不合适了。常见做法是加一个 ANP 解析函数让调度方只认 Agent 名字import json, time, urllib.request _anp_cache {} def resolve_agent(agent_name: str, ttl: int 60) - str: 根据 ANP 名字解析出 A2A endpoint带 TTL 缓存。 cached _anp_cache.get(agent_name) if cached and cached[expire_at] time.time(): return cached[endpoint] # 真实项目中这里查询内部 ANP 目录服务而不是读本地文件 with open(anp_records.json, r, encodingutf-8) as f: records json.load(f) record records.get(agent_name) if not record: raise KeyError(fagent {agent_name} not found in ANP directory) _anp_cache[agent_name] { endpoint: record[endpoint], expire_at: time.time() ttl } return record[endpoint]这个函数的关键参数是ttl它决定解析结果在本地缓存多久。Agent 地址变更频繁时把 TTL 调小但会增加对 ANP 目录的查询压力地址基本不变时 TTL 可以调大。正常情况下我会先设 60 秒等业务稳定后再逐步加长。这样调度代码永远只依赖ops-agent.main这样的名字而不是某个具体域名。4. MCP/A2A/ANP 联调避坑超时、握手失败与寻址冲突的排查清单这一章是我在实际联调里反复踩过的坑每条都按“现象 → 原因 → 解决”来写。你照着排查顺序走能省下大量抓瞎的时间。4.1 MCP Server 启动成功但 Client 连不上现象Server 进程正常启动日志也打了但客户端里的 MCP 工具列表一直是空的或者显示 disconnected。原因最常见的两个原因一是 Server 里用了print()往 stdout 打日志把 MCP 协议通道污染了二是 Server 在完成注册前就异常退出了比如 import 的依赖找不到进程闪退。解决先把所有print()改成logging并要求日志输出到文件或 stderr。接着在终端手动跑一次python ops_server.py观察退出码和报错。如果本地跑得好、但客户端里失败检查客户端配置里的command和args是否正确尤其是使用绝对路径还是相对路径。最后用官方调试命令mcp dev ops_server.py启动直接用调试面板看工具列表能立刻区分是 Server 问题还是 Client 配置问题。4.2 MCP Client 报 timed out after 30 seconds现象工具调用偶尔成功偶尔报错错误信息里出现timed out after 30 seconds之类的字样。在 Codex 等客户端里这个错误会明确提示去调整超时或任务模式。原因Client 对单次工具调用有默认超时限制一般是 30 秒。你的 MCP 工具如果是个慢查询、要等外部 API 返回很容易撞上这个上限。解决第一步把配置里的timeout从默认值调大比如 60 秒或 120 秒确认工具本身能正常完成。第二步把慢操作改成两段式MCP 工具只负责提交任务并返回任务 IDAgent 再通过另一个工具轮询结果。这样单次调用永远控制在几百毫秒内既绕开超时又不会阻塞 Agent 的推理循环。不要试图无限调大超时那只适合调试不适合生产。4.3 A2A 任务一直卡在 working现象message/send返回正常taskId 也拿到了但轮询tasks/get时状态永远停在 working永远到不了 completed。原因最常见的是服务端把任务丢进后台执行后执行逻辑崩溃了但异常没有被捕获任务状态也就没有更新还有一种情况是调用方根本没有实现轮询只发了一次消息就什么都不管了。解决在服务端给每个任务加超时保护超过 N 分钟强制置为 failed并记录失败原因。同时把执行体包进 try/except任何异常都回写任务状态。调用方则要实现轮询逻辑轮询间隔建议 1 到 2 秒最大重试次数根据任务时长估算。A2A 协议本身允许状态回退但你的代码必须保证每一个状态变更都有日志否则任务卡住时你连现场都找不到。4.4 ANP 解析结果漂移同一个名字指向两个地址现象同一个ops-agent.main有时候调用成功有时候连到一个不存在的服务报 connection refused。原因ANP 记录被覆盖时没有做所有权校验比如测试环境的 Agent 注册记录把生产环境的覆盖了或者旧记录没有按 TTL 过期新记录同时又存在导致不同调用方拿到不同结果。解决在 ANP 记录里增加签名字段注册时用 Agent 私钥签名查询方验签通过才使用。统一 TTL避免某些记录缓存过长。还有一条血泪经验在更新记录前先查一次当前 owner如果是别的团队在维护必须走变更流程不要直接覆盖。多环境场景下ANP 名字里带上环境后缀比如ops-agent.main和ops-agent.staging彻底隔离。4.5 一直报 method not found现象A2A 端点能收到请求但每次都返回method not found任务完全推不下去。原因这是一个典型的协议混用问题。有人把 MCP 的tools/call直接 POST 给了 A2A 的端点或者反过来把message/send发给了 MCP Server。两个协议虽然都用 JSON-RPC但方法名完全不同。解决先抓包看请求体里的 method 字段确认走的是哪个协议。MCP 服务端只认tools/list、tools/callA2A 服务端只认message/send、tasks/get。如果你在自己的业务网关上统一收流量按 method 前缀路由不要把两个协议的服务混在同一个函数里。顺手把收到的未知 method 打印出来能省掉很多抓包的时间。5. 落地演进路线从单 Agent 工具接入到企业级多 Agent 网络本地链路跑通以后下一个问题是怎么把它变成一个能稳定承载业务的系统。从单 Agent 到多 Agent不只是一次加两个协议那么简单还有工具复用、能力注册、权限审计、日志管理这些工程问题要处理。5.1 一个 MCP Server 服务多个 Agent如果你的 MCP Server 还在用 stdio 模式它就只能服务本地启动它的那一个 Agent。要让多个 Agent 共享同一组工具常见做法是把 MCP Server 改成 HTTP 模式部署if __name__ __main__: mcp.run(transportstreamable_http, host0.0.0.0, port9000)改成这个方式后所有 Agent 通过http://agent-tools.internal:9000/mcp这个 URL 连接同一个 MCP Server。好处是工具只有一份维护时只需要更新一个服务所有 Agent 立即生效。坏处是 HTTP 模式需要额外考虑鉴权、限流和网络策略不能像 stdio 那样只在本地悄悄跑。我见过很多团队把工具逻辑散落在各个 Agent 的私有目录里每个 Agent 一套脚本最后同一个监控工具被复制了三份参数还互相不一致。正确的做法是收口成独立的 MCP 工具服务所有 Agent 都从同一个服务拿工具能力。5.2 工具发现与能力注册让 Agent 自己找到可用工具Agent 数量多了以后每个 Agent 要连接哪些 MCP Server不能靠人肉写配置。更可靠的做法是做一个能力注册中心定期扫描所有 MCP Server 的工具列表存成索引Agent 启动时先查索引再建立连接import asyncio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client async def fetch_tools(command: str, args: list[str]) - list[dict]: server_params StdioServerParameters(commandcommand, argsargs) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() tools await session.list_tools() return [{name: t.name, description: t.description} for t in tools.tools] if __name__ __main__: tools asyncio.run(fetch_tools(python, [ops_server.py])) print(tools)这段代码在启动时连接 MCP Server调用list_tools拉取全部工具名和描述然后统一写入注册中心。这里要重点看session.initialize()MCP 客户端在调用工具前必须先完成初始化握手跳过这一步会出现连接状态错误。工具描述会原样进入注册中心所以 MCP Server 里每个工具的 docstring 都要写清楚用途、参数和典型场景这不只是给大模型看的也是在给注册中心建立可检索的能力目录。5.3 权限、审计与自定义日志管理MCP Server 暴露的工具并不都是无害的像重启服务、写数据库这类高权限工具必须做工具级白名单。我的惯例是每个工具函数开头先做三件事校验调用者身份、校验参数合法性、记录操作审计日志。import logging from logging.handlers import RotatingFileHandler logger logging.getLogger(mcp-audit) fh RotatingFileHandler(audit.log, maxBytes20_000_000, backupCount10) fh.setFormatter(logging.Formatter(%(asctime)s | %(levelname)s | %(message)s)) logger.addHandler(fh) def restart_service(service_name: str, requester: str unknown) - str: allowed {web, worker} if service_name not in allowed: logger.warning(denied restart | requester%s service%s, requester, service_name) return denied logger.info(restart accepted | requester%s service%s, requester, service_name) # 执行重启逻辑 return f{service_name} restarting这套写法的核心是“拒绝也要记录”。在实际生产里真正有价值的安全信息往往是那些被拦下来的请求它告诉你有人在尝试调用你没打算开放的能力。日志文件建议按天切分并用logger.info(字段值)这种结构化的方式打点而不是打一段自然语言。这样后续接入日志采集系统时可以直接按service或requester维度聚合检索。5.4 别把 MCP 当 RAG 用不少刚接触 MCP 的人会问我已经可以把文档地址传给 MCP 工具去读取了是不是就不需要 RAG 了这是两件事。MCP 解决的是“Agent 能调用什么动作”RAG 解决的是“Agent 从哪里拿知识”。动作和知识必须分开场景适合 MCP适合 RAG查数据库里的用户订单是通过 SQL 工具不推荐会幻觉从几百篇文档里找相似段落不推荐逐篇读太慢是向量检索重启故障服务是直接调用运维工具否回答最新版内部规范否是先检索再回答正确路径是先通过 RAG 定位到相关知识片段再把知识片段作为参数交给 MCP 工具去执行。比如“查一下最近三天登录失败的账号”先用 RAG 或普通搜索确定日志位置再调 MCP 工具读文件。反过来先调一堆工具再翻文档效率很低也容易让模型在工具返回的大量原始数据里迷失焦点。6. 验证与进阶技巧一个请求跨了三层协议怎么定位到底卡在哪本地能跑通不代表链路稳。多 Agent 场景里一次业务请求往往要经历 ANP 解析、A2A 消息投递、MCP 工具调用三个环节任何一个环节慢都会拖垮整体延迟。我的验证习惯是从入口请求开始生成一个 traceId把它塞进每一层协议的消息体里。A2A 消息的 params 对象里可以加自定义字段MCP 工具调用时也可以通过参数传入 traceId。然后在 MCP Server 日志、A2A 任务对象、ANP 解析记录里同时打印这个 traceId。日志格式统一成 JSON Lines后续直接用jq按 traceId 过滤cat mcp_ops.log | jq select(.trace_id trace-20250101-001)这样你能清楚看到时间线ANP 解析花了多少毫秒A2A 消息到达目标 Agent 后排队多久MCP 工具实际执行了几秒。我这里有个真实的教训之前有一个任务经常超时界面只显示“Agent 调用失败”我以为是 A2A 协议不稳定反复调了三四天握手逻辑。后来加了 traceId 一通查才发现瓶颈在 MCP 工具内部的数据库慢查询跟协议一点关系都没有。协议层的问题通常很显眼真正难抓的是那些藏在工具实现里的延迟。所以我的最后一条建议是不要在一开始就追求协议全家桶。先把 MCP 用明白再逐步加 A2AANP 留到实在需要管理大量动态端点时再上。协议只是骨架真正决定系统好不好用的是每一个工具的质量和日志的可观测性。希望这些思路和踩坑记录能帮你少走一段弯路。本文还有配套的精品资源点击获取
