在MCP上实现Agent间通信：完整技术指南

MCP已显著超越其最初"为LLM提供上下文"的目标。随着可恢复流、请求输入、采样和通知等增强功能的加入，MCP现在为构建复杂的Agent间通信系统提供了坚实基础。

Agent/工具误解

随着更多开发者探索具有Agent行为（长时间运行、执行中可能需要额外输入等）的工具，一个常见的误解是MCP不适合构建Agent，主要是因为其早期工具示例专注于简单的请求-响应模式。

这种认知已经过时。MCP规范在过去几个月中已显著增强，缩小了构建长时间运行Agent行为的差距：

流式和部分结果：执行期间的实时进度更新
可恢复性：客户端可以在断开连接后重新连接并继续
持久性：结果在服务器重启后仍然存在
多轮交互：通过请求输入和采样在执行中交互式输入

什么使MCP工具具有"Agent"特性？

我们将Agent定义为能够在较长时间内自主运行、处理可能需要基于实时反馈进行多次交互或调整的复杂任务的实体。为了支持这些行为，MCP工具需要四个关键基础设施能力：

1. 流式和部分结果

Agent需要实时进度更新机制，让用户了解情况，并允许系统基于部分输出进行调整和反应。

MCP支持：通过进度通知实时流式传输状态更新

2. 可恢复性

长时间运行的Agent受益于在网络中断期间保持任务连续性，允许客户端重新连接并从断点恢复。

MCP支持：StreamableHTTP传输支持会话恢复和消息重传

3. 持久性

长时间运行的Agent需要持久状态，在服务器重启后仍然存在，并支持跨会话的进度跟踪。

MCP支持：资源链接 - 工具可以创建资源并立即返回其链接，然后在后台继续处理

4. 多轮交互

Agent可能需要在执行中获得额外输入 - 用于关键决策的人工澄清或批准，以及用于复杂子任务的AI生成内容。

MCP支持：通过请求输入和采样完全支持这些交互

在MCP上实现长时间运行Agent - 代码概述

我们提供了一个代码仓库，包含使用MCP Python SDK和StreamableHTTP传输实现长时间运行Agent的完整实现。

我们的代码示例演示了主机应用程序如何连接到具有多个工具的服务器：

旅行Agent - 通过请求输入模拟带有价格确认的旅行预订服务
研究Agent - 通过采样执行带有AI辅助摘要的研究任务

两个Agent都演示了实时进度更新、交互式确认和完整的会话恢复能力。

关键实现概念

流式和进度更新 - 实时任务状态

服务器实现：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10


# 从server/server.py - 旅行Agent发送进度更新
for i, step in enumerate(steps):
    await ctx.session.send_progress_notification(
        progress_token=ctx.request_id,
        progress=i * 25,
        total=100,
        message=step, 
        related_request_id=str(ctx.request_id)   
    )
    await anyio.sleep(2)  # 模拟工作

请求输入 - 请求用户输入

服务器实现：

1
2
3
4
5
6


# 从server/server.py - 旅行Agent请求价格确认
elicit_result = await ctx.session.elicit(
  message=f"请确认您前往{destination}旅行的预估价格$1200",
  requestedSchema=PriceConfirmationSchema.model_json_schema(),
  related_request_id=ctx.request_id,
)

采样 - 请求AI协助

服务器实现：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11


# 从server/server.py - 研究Agent请求AI摘要
sampling_result = await ctx.session.create_message(
  messages=[
    SamplingMessage(
      role="user",
      content=TextContent(type="text", text=f"请总结关于{topic}研究的关键发现")
    )
  ],
  max_tokens=100,
  related_request_id=ctx.request_id,
)

可恢复性 - 跨断开连接的会话连续性

事件存储实现：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12


# 从server/event_store.py - 简单的内存事件存储
class SimpleEventStore(EventStore):
  def __init__(self):
    self._events: list[tuple[StreamId, EventId, JSONRPCMessage]] = []
    self._event_id_counter = 0

  async def store_event(self, stream_id: StreamId, message: JSONRPCMessage) -> EventId:
    """存储事件并返回其ID"""
    self._event_id_counter += 1
    event_id = str(self._event_id_counter)
    self._events.append((stream_id, event_id, message))
    return event_id

扩展到MCP上的多Agent通信

架构说明：我们的实现已经演示了Agent间通信。主机应用程序充当"编排器Agent"，与用户交互并将请求路由到MCP服务器上的专业Agent。

虽然我们的示例为简单起见连接到单个服务器，但相同的编排器Agent可以跨多个MCP服务器协调任务，每个服务器暴露不同的专业Agent。

要将主机Agent扩展到远程MCP Agent模式到多个服务器，可以增强主机Agent：

智能任务分解
多服务器协调
任务状态管理
用户上下文保持
弹性和重试
结果合成

结论

MCP的增强能力 - 资源通知、请求输入/采样、可恢复流和持久资源 - 在保持协议简单性的同时实现了复杂的Agent间交互。

总体而言，MCP协议规范正在快速发展；鼓励读者查看官方文档网站以获取最新更新。