企业官网建设流程全解析

1. 从两个缩写说起：MCP 与 A2A 到底在解决什么问题

第一次看到 MCP 和 A2A 这两个词放在一起，很多人会本能地以为是某种硬件通信协议，比如 CAN、Modbus、IIC 那一类跑在物理总线上的东西。实际上它们都属于应用层的软件协议，解决的是"不同程序之间怎么把话说清楚"的问题，只不过一个面向的是模型和工具之间的连接，另一个面向的是智能体与智能体之间的协作。

MCP 全称 Model Context Protocol，核心目标是给大模型应用提供一个标准化的方式去访问外部资源——文件、数据库、API、命令行工具都算。在它出现之前，每接一个工具就要写一套适配代码，工具一多，维护成本直接爆炸。MCP 把这件事抽象成"客户端-服务端"模型，工具方只需要实现一个 MCP Server，任何支持 MCP 的客户端都能直接调用，省掉了大量重复对接的工作。

A2A 全称 Agent to Agent，关注的是另一个层面：当你有多个智能体，各自负责不同任务时，它们之间怎么协商、怎么传递任务、怎么汇报结果。MCP 解决的是"智能体怎么用工具"，A2A 解决的是"智能体怎么找智能体帮忙"。两者不是竞争关系，而是互补关系，一个向内连接资源，一个向外连接同伴。

这篇内容适合谁看？如果你正在做 AI 应用集成、想让自己的系统接入大模型能力、或者手上有多个自动化流程需要串起来，那这两个协议值得花时间搞清楚。下面我会从设计思路、协议细节、实操落地到踩坑排查，把整套东西拆开讲一遍，尽量做到看完就能动手。

2. 协议整体设计与选型思路拆解

2.1 为什么是 JSON-RPC 作为底座

MCP 和 A2A 在传输层都选择了 JSON-RPC 2.0 作为消息格式，这个选择不是随便定的。JSON-RPC 的结构极其简单，一个请求就是jsonrpc、method、params、id四个字段，响应就是result或error二选一。这种极简结构带来的好处是：任何语言都能在半小时内实现一个可用的编解码器，不需要引入厚重的框架。

对比一下其他选项就明白了。如果用 gRPC，你得维护 proto 文件、处理代码生成、还要考虑 HTTP/2 的兼容性，对于工具调用这种请求-响应模式来说太重了。如果用纯 REST，虽然通用，但缺少统一的错误码规范和请求 ID 机制，做双向通信和异步任务时会很别扭。JSON-RPC 刚好卡在中间：比 REST 规范，比 gRPC 轻量。

提示：JSON-RPC 的id字段是关联请求和响应的关键，异步场景下必须保证同一会话内 id 不重复，否则响应会串线。

2.2 Streamable HTTP 解决了什么痛点

早期 MCP 的远程传输用的是 HTTP + SSE 的组合，服务端通过 SSE 长连接往客户端推消息。这套方案在本地跑没问题，但一旦部署到有负载均衡、有网关的环境里就出问题：SSE 连接容易被中间层掐断，重连之后上下文丢失，多实例部署时还会出现请求打到 A 实例、推送却从 B 实例发出的情况。

Streamable HTTP 的思路是把"流"和"请求"统一到同一个 HTTP 端点上。客户端发一个 POST 请求，服务端可以选择直接返回一个 JSON 响应，也可以返回一个text/event-stream的流式响应，在流里持续推送多条消息。关键在于这个流是可断可续的，服务端可以给每条消息打上事件 ID，客户端断线后用Last-Event-ID头重新发起请求，就能从断点继续接收。

这个设计对部署非常友好。因为所有交互都走标准 HTTP，负载均衡器不需要做任何特殊配置，会话状态可以通过事件 ID 在服务端重建，多实例部署时只要共享会话存储就能正常工作。实测下来，这套机制在容器化环境里的稳定性比老的 SSE 方案高出一个档次。

2.3 MCP 与 A2A 的职责边界

很多人会问：既然 MCP 已经能让智能体调用工具了，为什么还需要 A2A？答案在于抽象层级不同。MCP 里的工具是"无状态的能力单元"，比如查天气、读文件、执行 SQL，它不关心谁在调用，也不维护跨请求的上下文。而 A2A 里的智能体是"有状态的协作方"，它可能记得之前的对话、有自己的任务队列、甚至需要反过来向调用方请求补充信息。

举个实际场景：你要做一个自动化的报表系统。用 MCP，你可以让模型调用"查询数据库"和"生成图表"两个工具，这是工具层面的组合。但如果报表生成需要经过"数据校验智能体"和"格式排版智能体"两道工序，两者之间要传递中间结果、要处理校验失败后的重试，这就属于 A2A 的范畴了。

选型上的经验是：单一职责、无状态、调用即返回的能力，用 MCP 封装成工具；需要多轮交互、有内部状态、可能反向提问的，用 A2A 定义成智能体。两者可以在同一个系统里共存，MCP Server 也可以被 A2A 智能体当作工具来调用。

3. 核心细节解析与实操要点

3.1 MCP 的三类核心原语

MCP 协议里定义了三种服务端能力，理解它们的区别是上手的关键。

Tools（工具）是最常用的一类，代表可以被模型主动调用的函数。每个工具需要提供名称、描述和 JSON Schema 格式的参数定义。模型根据描述决定要不要调用、传什么参数。这里有个容易踩的坑：描述写得太模糊，模型就不知道该在什么场景下用；描述写得太长，又会挤占上下文窗口。我的经验是描述控制在两句话以内，第一句说清楚"做什么"，第二句说清楚"什么时候用"。

Resources（资源）代表可以被读取的数据，比如文件内容、数据库记录、配置项。和工具不同，资源是被动的，通常由客户端或用户决定要不要加载，而不是模型自主决定。资源用 URI 标识，支持file://、db://这类自定义 scheme。

Prompts（提示模板）是预定义的提示词模板，服务端可以提供一组参数化的 prompt，客户端调用时填入参数就能得到完整的提示词。这个能力在需要固定输出格式的场景下特别有用，比如"代码审查"模板、"周报生成"模板。

3.2 工具定义的 Schema 设计要点

工具的参数 Schema 直接决定了模型能不能正确调用。我见过太多因为 Schema 设计不当导致调用失败的案例，这里总结几条硬性经验。

参数类型尽量用基础类型，string、number、boolean、array这几种就够了。避免用嵌套很深的 object，模型在深层嵌套上出错率明显更高。如果确实需要复杂结构，拆成多个扁平参数，在服务端再组装。

每个参数都要写description，而且要写清楚格式要求。比如日期参数，光写"日期"不够，要写"ISO 8601 格式，例如 2024-01-15"。枚举参数要把所有可选值列在enum里，不要指望模型猜。

必填参数用required数组明确标注。可选参数要给合理的默认值，并且在描述里说明默认行为。我踩过的一个坑是：某个可选参数没写默认值，模型有时传有时不传，导致服务端逻辑分支爆炸，后来统一在服务端做了默认值兜底才稳定下来。

{ "name": "query_orders", "description": "查询指定时间范围内的订单。用于用户询问订单情况时。", "inputSchema": { "type": "object", "properties": { "start_date": { "type": "string", "description": "开始日期，ISO 8601 格式，例如 2024-01-15" }, "end_date": { "type": "string", "description": "结束日期，ISO 8601 格式，例如 2024-01-20" }, "status": { "type": "string", "enum": ["pending", "paid", "shipped", "completed"], "description": "订单状态筛选，不传则返回全部状态" } }, "required": ["start_date", "end_date"] } }

3.3 A2A 的任务生命周期

A2A 的核心是任务（Task）这个概念。一个任务从创建到结束会经历若干状态：submitted、working、input-required、completed、failed、canceled。理解状态流转是排查问题的前提。

input-required这个状态是 A2A 区别于普通 RPC 的关键。当智能体执行到一半发现信息不足时，它可以把任务置为这个状态，并附带需要补充的信息说明。调用方看到这个状态后，补充信息再继续推进任务。这种"反向提问"的能力让智能体之间的协作更接近人类之间的沟通。

任务的状态变更通过流式事件推送。每次状态变化都会产生一个事件，事件里包含任务 ID、当前状态和相关的消息内容。调用方根据事件类型决定下一步动作。这里要注意：事件可能乱序到达，处理逻辑必须基于任务 ID 做幂等，不能假设事件严格有序。

3.4 会话与上下文管理

MCP 的会话通过Mcp-Session-Id头来标识。客户端首次连接时服务端返回一个会话 ID，后续请求都带上这个头。服务端根据会话 ID 维护上下文，比如已加载的资源、已建立的连接。

A2A 的上下文管理更复杂一些，因为涉及多个智能体。每个任务有独立的上下文，任务之间通过contextId关联。同一个 context 下的多个任务可以共享历史消息，这样智能体在处理新任务时能参考之前的交互。

实操中一个常见问题是会话过期。服务端通常会设置会话超时时间，超时后会话 ID 失效，客户端需要重新初始化。如果客户端没处理好这个逻辑，就会出现"突然所有请求都返回会话无效"的情况。我的做法是在客户端加一层自动重连：捕获会话失效错误后，自动重新初始化并重放当前请求。

4. 实操过程与核心环节实现

4.1 搭建一个最小可用的 MCP Server

先从一个最简单的 MCP Server 开始，实现一个查询当前时间的工具。用 Python 的官方 SDK 来做，整个代码不到五十行。

from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import Tool, TextContent import datetime app = Server("time-server") @app.list_tools() async def list_tools(): return [ Tool( name="get_current_time", description="获取当前时间。用于用户询问现在几点时。", inputSchema={ "type": "object", "properties": { "timezone": { "type": "string", "description": "时区名称，例如 Asia/Shanghai，默认 UTC" } } } ) ] @app.call_tool() async def call_tool(name: str, arguments: dict): if name == "get_current_time": tz = arguments.get("timezone", "UTC") now = datetime.datetime.now() return [TextContent(type="text", text=f"当前时间：{now.isoformat()}，时区：{tz}")] async def main(): async with stdio_server() as (read, write): await app.run(read, write, app.create_initialization_options()) if __name__ == "__main__": import asyncio asyncio.run(main())

这段代码用的是 stdio 传输，也就是通过标准输入输出通信，适合本地进程调用。如果要部署成远程服务，需要换成 Streamable HTTP 传输。

4.2 切换到 Streamable HTTP 传输

远程部署时，把传输层换成 HTTP。下面是基于 Starlette 的实现骨架。

from mcp.server.streamable_http import StreamableHTTPServerTransport from starlette.applications import Starlette from starlette.routing import Route transport = StreamableHTTPServerTransport( mcp_session_id=None, is_json_response_enabled=True ) async def handle_mcp(request): return await transport.handle_request(request) app = Starlette(routes=[ Route("/mcp", handle_mcp, methods=["POST", "GET", "DELETE"]) ])

这里有几个关键点。is_json_response_enabled设为 True 时，服务端对简单请求直接返回 JSON，对需要流式的请求返回 SSE 流。路由要同时支持 POST、GET、DELETE 三种方法：POST 用于发送请求，GET 用于建立流式连接，DELETE 用于主动关闭会话。

部署到容器环境时，要注意会话存储的问题。如果服务端是多实例的，会话状态必须放在共享存储里，比如 Redis。否则客户端第一次请求打到实例 A，第二次打到实例 B，B 找不到会话就会报错。我一般会在会话存储里存三样东西：会话 ID、已初始化的能力列表、最后活跃时间。

4.3 实现一个 A2A 智能体

A2A 智能体的实现比 MCP Server 复杂一些，因为要处理任务状态流转。下面是一个处理"文本摘要"任务的智能体骨架。

from a2a.server import A2AServer from a2a.types import Task, TaskState, Message class SummaryAgent(A2AServer): async def on_task(self, task: Task): text = task.message.parts[0].text if len(text) < 50: await self.update_task( task.id, TaskState.INPUT_REQUIRED, Message(text="文本太短，请提供至少 50 字的文本") ) return await self.update_task(task.id, TaskState.WORKING) summary = await self.summarize(text) await self.update_task( task.id, TaskState.COMPLETED, Message(text=summary) ) async def summarize(self, text: str) -> str: # 实际摘要逻辑 return text[:100] + "..."

这个例子里，智能体先检查输入长度，不够就返回input-required状态要求补充。够的话进入working状态，处理完置为completed。调用方通过流式事件感知这些状态变化。

4.4 客户端如何同时对接 MCP 和 A2A

实际系统里，客户端往往需要同时管理 MCP 连接和 A2A 连接。我的做法是抽象一个统一的"能力注册表"，把 MCP 工具和 A2A 智能体都注册进去，对上层暴露统一的调用接口。

class CapabilityRegistry: def __init__(self): self.mcp_clients = {} self.a2a_clients = {} async def register_mcp(self, name, url): client = await MCPClient.connect(url) tools = await client.list_tools() self.mcp_clients[name] = {"client": client, "tools": tools} async def register_a2a(self, name, url): client = await A2AClient.connect(url) card = await client.get_agent_card() self.a2a_clients[name] = {"client": client, "card": card} async def invoke(self, target, method, params): if target in self.mcp_clients: return await self.mcp_clients[target]["client"].call_tool(method, params) elif target in self.a2a_clients: return await self.a2a_clients[target]["client"].send_task(params)

这层抽象的好处是上层业务代码不用关心底层是 MCP 还是 A2A，统一按"目标 + 方法 + 参数"的方式调用。新增能力时只需要注册，不用改业务逻辑。

4.5 参数计算与超时设置

超时设置是实操中最容易被忽视的环节。MCP 工具调用和 A2A 任务执行的耗时差异很大，不能用一个统一的超时值。

我的经验值是这样的：MCP 工具调用超时设 30 秒，因为大部分工具是查询类操作，超过 30 秒基本可以判定有问题。A2A 任务超时设 5 分钟，因为涉及多轮交互和复杂处理。流式连接的空闲超时设 60 秒，超过这个时间没有事件推送就认为连接已死，主动重连。

重连策略用指数退避：第一次失败等 1 秒，第二次等 2 秒，第三次等 4 秒，最多退避到 30 秒。连续失败 5 次后停止重连并告警。这个策略在实测中能有效应对服务端短暂重启的情况，又不会在服务端彻底挂掉时无限重试。

5. 常见问题与排查技巧实录

5.1 工具调用返回"方法不存在"

这是最常见的问题，通常有三个原因。第一是工具名拼写不一致，客户端注册时用的名字和服务端定义的名字大小写不同。第二是会话未初始化，服务端还没收到initialize请求就收到了tools/call。第三是服务端版本不匹配，客户端按新协议发请求，服务端还是老版本。

排查顺序：先看服务端日志有没有收到initialize，再看工具列表里有没有目标工具，最后对比客户端和服务端的协议版本号。我一般会在客户端启动时打印一次完整的工具列表，出问题时对照这个列表就能快速定位。

5.2 流式连接频繁断开

流式连接断开的原因比较多，按概率排序：中间层网关的空闲超时、服务端心跳没发、客户端没处理重连。

网关空闲超时是最常见的。很多云厂商的负载均衡默认 60 秒无数据就断连接。解决办法是服务端定期发送心跳事件，比如每 30 秒发一个空的事件。这样连接始终有数据流动，不会被判定为空闲。

客户端侧要正确处理Last-Event-ID。断线重连时带上最后收到的事件 ID，服务端就能从断点继续推送。如果客户端没实现这个逻辑，重连后要么丢消息，要么重复处理。

5.3 A2A 任务卡在 working 状态

任务一直停在working不结束，通常是智能体内部抛了异常但没捕获，导致状态没更新。排查方法是看智能体的日志，找有没有未处理的异常。

另一个可能是任务处理时间确实很长，超过了客户端的超时设置。这种情况要么调大超时，要么让智能体定期发送进度事件，让客户端知道任务还在推进。

还有一种隐蔽的情况：智能体在处理过程中调用了另一个 A2A 智能体，但那个智能体没响应，导致当前智能体一直等待。这种级联等待很难排查，我的做法是给每个智能体设置内部调用超时，超时后主动失败并上报，避免无限等待。

5.4 常见问题速查表

5.5 几个独家避坑技巧

第一个技巧：在开发阶段把 MCP 和 A2A 的原始报文都打到日志里。JSON-RPC 的报文不大，全量记录对性能影响很小，但排查问题时能省下大量时间。我一般会用一个环境变量控制开关，生产环境关掉，出问题时临时打开。

第二个技巧：给每个工具和智能体加一个"健康检查"方法。客户端启动时先调一遍健康检查，确认对端可用再注册。这样能避免注册了一堆不可用的能力，调用时才发现问题。

第三个技巧：A2A 任务的contextId要设计得有语义。不要用随机 UUID，用"业务类型-日期-序号"这种格式，比如report-20240115-001。这样排查问题时一眼就能看出这个任务属于哪个业务、什么时候创建的。

第四个技巧：MCP 工具的返回值尽量结构化。返回纯文本虽然简单，但客户端要解析就很麻烦。返回 JSON 字符串，客户端直接反序列化，省去正则匹配的麻烦。如果确实需要返回文本，在 Schema 里明确说明格式。

6. 协议组合使用的进阶思路

6.1 用 MCP 封装底层能力，用 A2A 编排业务流程

一个比较成熟的架构是分两层：底层用 MCP 把各种原子能力封装成工具，上层用 A2A 定义业务流程智能体。业务智能体不直接调用底层 API，而是通过 MCP 工具来操作数据。

这样分层的好处是职责清晰。底层能力的变化只影响 MCP Server，业务流程的变化只影响 A2A 智能体，两者解耦。新增一个数据源时，只需要加一个 MCP Server，所有业务智能体都能用上。

6.2 智能体之间的任务分解与结果聚合

复杂任务往往需要分解成多个子任务，分给不同的智能体处理，最后聚合结果。A2A 本身不提供任务分解能力，这部分逻辑要自己实现。

我的做法是定义一个"协调者"智能体，它接收大任务后，根据任务类型拆成子任务，通过 A2A 分发给对应的执行智能体，收集所有结果后聚合返回。协调者本身也是一个 A2A 智能体，对外表现和普通智能体一样。

任务分解的粒度要把握好。拆得太细，协调开销大；拆得太粗，单个智能体负担重。经验值是每个子任务的处理时间控制在 10 到 60 秒之间，这个区间内并行收益比较明显，协调开销也可控。

6.3 错误传播与降级策略

多智能体协作时，一个环节出错可能影响整个流程。错误传播的处理策略有三种：快速失败、重试、降级。

快速失败适合关键路径上的错误，比如数据校验不通过，直接终止整个流程并返回错误。重试适合临时性错误，比如网络抖动，重试几次通常能成功。降级适合非关键环节，比如某个增强功能不可用，跳过它继续走主流程。

实现上，我会给每个子任务标注"关键性"和"可重试性"两个属性。协调者根据这两个属性决定错误处理策略。关键且不可重试的，直接失败；非关键或可重试的，按策略处理。

7. 我在实际项目中的几点体会

这套协议组合用下来，最大的感受是"标准化带来的复用价值远超预期"。以前每接一个新系统都要写适配层，现在只要对方实现了 MCP 或 A2A，对接工作量能减少七成以上。特别是 MCP 的工具体系，一旦积累了一批常用工具，新项目启动时直接复用，效率提升非常明显。

另一个体会是协议版本管理要提前规划。MCP 和 A2A 都还在演进，字段和语义可能变化。我的做法是在客户端和服务端都记录协议版本，握手时校验版本兼容性，不兼容就明确报错，而不是带着隐患继续跑。这样虽然前期麻烦一点，但能避免很多诡异的问题。

最后分享一个小技巧：调试 MCP 和 A2A 时，用一个通用的"回显"工具或智能体作为参照。先确认回显能正常工作，再排查具体业务逻辑。这样能把"协议层问题"和"业务层问题"快速区分开，排查效率能提高不少。很多时候折腾半天，最后发现是协议层没通，而不是业务代码写错了。