前言

过去我们做软件系统集成，常见的是 REST API、RPC、WebSocket、消息队列这些技术。但进入 AI Agent 时代之后，我们会发现新的问题出现了：

不是系统之间简单地调用接口，而是 AI 客户端、IDE、Agent、外部工具、聊天平台之间需要协同工作。

例如：

• Claude Desktop 想调用你本地写的“查公司内部 Wiki”的工具。
• Cursor / VS Code 想让一个 coding agent 接管项目重构任务。
• 一个企业 AI 助手想同时接入微信、Slack、Discord、Telegram。
• Agent 在 IDE 中修改文件时，需要向客户端请求权限、返回进度、支持取消任务。
• AI 模型想知道有哪些工具可以调用、工具参数是什么、返回结果是什么。

这些场景背后，本质上都需要一套“通信协议”。

这篇文章围绕几个关键协议和架构模式展开：

1. JSON-RPC 2.0：底层消息格式。
2. stdio Transport：本地进程间通信方式。
3. LSP：IDE 与语言服务器之间的经典协议。
4. MCP：AI 客户端发现和调用外部工具的协议。
5. ACP：IDE 与 AI coding agent 之间的协议。
6. Gateway Plugin：把 Agent 接入微信、Slack、Discord 等聊天平台的插件架构。

如果用一句话概括它们的关系：

JSON-RPC 是消息格式，stdio/HTTP/SSE 是传输通道，LSP 是 IDE 时代的语言能力协议，MCP 是 AI 调用外部工具的协议，ACP 是 IDE 驱动 Agent 的协议，Gateway Plugin 是 Agent 接入多聊天平台的适配层。

---

一、先理解 JSON-RPC：这些协议的共同底座

无论是 LSP、MCP 还是 ACP，它们都大量借鉴或直接使用了 JSON-RPC 2.0。

JSON-RPC 是一种非常轻量的远程调用协议。它不像 REST 那样围绕 URL 和 HTTP Method 设计，而是围绕“方法名 + 参数 + 请求 ID”设计。

JSON-RPC 2.0 官方规范中定义了 Request、Response、Notification、Error 等核心结构；其中 Notification 是没有 id 的请求，服务端不需要返回响应。(jsonrpc.org)

1.1 JSON-RPC 请求格式

一个最简单的 JSON-RPC 请求长这样：

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/list",
  "params": {}
}

字段含义：

字段	含义
jsonrpc	协议版本，固定为 "2.0"
id	请求 ID，用来匹配响应
method	要调用的方法名
params	方法参数

服务端响应：

{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "tools": []
  }
}

如果出错，则返回：

{
  "jsonrpc": "2.0",
  "id": 1,
  "error": {
    "code": -32601,
    "message": "Method not found"
  }
}

1.2 Notification：不需要响应的消息

JSON-RPC 中还有一种特殊消息，叫 Notification。

它没有 id：

{
  "jsonrpc": "2.0",
  "method": "notifications/initialized",
  "params": {}
}

因为没有 id，服务端就不需要返回响应。

这类消息很适合用于：

• 初始化完成通知
• 进度更新
• 日志推送
• 状态变更
• 文件变化通知

JSON-RPC 2.0 规范明确说明，没有 id 的 Request 就是 Notification，服务端不能回复 Notification。(jsonrpc.org)

1.3 为什么这些协议喜欢用 JSON-RPC？

因为它非常适合本地工具、IDE、Agent 之间通信。

它的优点是：

1. 简单：一个 JSON 对象就能表达一次调用。
2. 语言无关：Python、TypeScript、Go、Rust、Java 都能轻松实现。
3. 传输无关：可以跑在 stdio、WebSocket、HTTP、TCP 上。
4. 天然支持双向通信：客户端可以请求服务端，服务端也可以发送通知。
5. 适合长生命周期进程：例如 IDE 启动一个 language server，持续通信。

这也是为什么 LSP、MCP、ACP 都和 JSON-RPC 有很深的关系。

---

二、再理解 stdio：为什么这些协议喜欢用标准输入输出？

很多人第一次看到 MCP 或 ACP 的时候，会疑惑：

为什么不用 HTTP，而是用 stdio？

stdio 指的是：

• stdin：标准输入
• stdout：标准输出
• stderr：标准错误

也就是说，客户端可以启动一个子进程，然后通过这个子进程的输入输出流通信。

例如：

Claude Desktop / Cursor / IDE
        |
        | 启动子进程
        v
python mcp_server.py
        |
        | stdin / stdout
        v
JSON-RPC 消息

2.1 stdio 的优势

1. 本地部署简单

不需要开放端口。

例如你写一个 MCP Server：

python wiki_mcp_server.py

AI 客户端直接启动这个进程即可。

不用考虑：

• 端口冲突
• 防火墙
• HTTPS 证书
• 反向代理
• 外网访问

2. 安全边界更清晰

stdio 通信通常发生在本机父子进程之间。

相比暴露 HTTP 服务，它的攻击面更小。

3. 非常适合 IDE / 桌面客户端

LSP 早就大量采用类似模式：

VS Code 启动 pyright / rust-analyzer / gopls
然后通过 JSON-RPC 通信

ACP 也借鉴了类似思想。ACP 官方相关文档说明，它是基于 JSON-RPC 的协议，客户端通常会以子进程方式启动 agent，并通过 stdio 通信，但协议本身也可以适配其他双向流。(docs.rs)

2.2 stdio 的缺点

stdio 也不是万能的。

它不适合：

• 多客户端共享同一个服务
• 跨机器访问
• 云端部署
• 需要负载均衡的服务
• 需要浏览器直接访问的服务

这时就更适合 HTTP、SSE、WebSocket 或 Streamable HTTP。

MCP 当前规范中也明确提到，MCP 使用 JSON-RPC 编码消息，并定义了 stdio 和 Streamable HTTP 两类标准传输机制。(模型上下文协议)

---

三、LSP：IDE 协议设计的经典范式

在讲 MCP 和 ACP 之前，必须先讲 LSP。

LSP 全称是 Language Server Protocol，即语言服务器协议。

它解决的问题是：

不同编辑器都需要代码补全、跳转定义、悬浮文档、诊断报错，但如果每个语言都给每个编辑器单独适配一遍，成本会爆炸。

以前的情况是：

Python 插件需要适配 VS Code
Python 插件需要适配 Vim
Python 插件需要适配 Emacs
Python 插件需要适配 JetBrains

Go 插件也要适配 VS Code
Go 插件也要适配 Vim
Go 插件也要适配 Emacs
……

这就是 N × M 的集成问题。

LSP 的思路是：

编辑器只实现 LSP Client，语言能力只实现 LSP Server，中间用统一协议通信。

VS Code / Vim / Emacs / Cursor
             |
             | LSP
             v
Python Language Server / Go Language Server / Rust Analyzer

LSP 官方规范描述的是编辑器和语言服务器之间的通信协议，当前稳定规范是 3.17。(GitHub Microsoft)

3.1 LSP 的核心流程

典型 LSP 流程：

1. initialize
2. initialized
3. textDocument/didOpen
4. textDocument/didChange
5. textDocument/completion
6. textDocument/hover
7. textDocument/definition
8. textDocument/publishDiagnostics

例如编辑器打开一个 Python 文件：

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "initialize",
  "params": {
    "processId": 12345,
    "rootUri": "file:///project",
    "capabilities": {}
  }
}

语言服务器返回自己支持的能力：

{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "capabilities": {
      "hoverProvider": true,
      "definitionProvider": true,
      "completionProvider": {
        "resolveProvider": true
      }
    }
  }
}

之后编辑器就知道：

这个语言服务器支持 hover、definition、completion。

3.2 LSP 的价值

LSP 最核心的价值是解耦：

编辑器不需要懂 Python / Go / Rust 的语义
语言服务器不需要关心 VS Code / Vim / Emacs 的 UI

编辑器只负责展示：

• 补全列表
• 错误波浪线
• 跳转结果
• 悬浮文档

语言服务器只负责计算：

• 代码语义
• 类型信息
• 诊断错误
• 定义位置
• 引用位置

这就是协议的力量。

LSP 通过统一协议把“编辑器”和“语言智能”解耦，减少了不同编辑器重复实现语言能力的成本。相关介绍也强调，LSP 让语言相关能力可以实现一次，并在多个支持 LSP 的编辑器中复用。(nabeelvalley.co.za)

3.3 LSP 对 AI Agent 协议的启发

LSP 的成功给 MCP / ACP 很大启发。

因为 AI Agent 时代也遇到了类似问题：

Claude Desktop 想接各种工具
Cursor 想接各种 agent
VS Code 想接各种 coding agent
聊天平台想接各种 AI 后端

如果每个客户端都和每个工具、每个 agent 单独适配，就会再次变成 N × M 的集成灾难。

所以我们需要新的标准协议：

MCP：统一 AI 调用工具的方式
ACP：统一 IDE 调用 Agent 的方式
Gateway Plugin：统一聊天平台接入 Agent 的方式

---

四、MCP：让 AI 客户端发现并调用外部工具

MCP 全称是 Model Context Protocol。

它解决的问题是：

AI 客户端如何标准化地发现外部工具、读取外部资源、调用外部能力？

MCP 官方文档中将 tools 描述为允许模型与外部系统交互的机制，例如查询数据库、调用 API 或执行计算；每个 tool 有唯一名称和描述其参数的 metadata/schema。(模型上下文协议)

4.1 MCP 的典型场景

假设你写了一个公司内部 Wiki 查询服务：

wiki_search(query: str) -> str

你希望 Claude Desktop、Cursor 或其他 AI 客户端能够调用它。

没有 MCP 时，你可能要为每个客户端单独写插件。

有 MCP 后，你可以写一个 MCP Server：

AI Client
   |
   | MCP
   v
Wiki MCP Server
   |
   v
公司内部 Wiki / 知识库 / 数据库

AI 客户端只需要知道：

1. 这个 MCP Server 暴露了哪些工具？
2. 每个工具叫什么？
3. 参数 schema 是什么？
4. 如何调用？
5. 返回结果是什么？

4.2 MCP 的核心角色

MCP 中通常有三个角色：

MCP Host：AI 应用本体，例如 Claude Desktop、Cursor
MCP Client：Host 内部负责协议通信的客户端
MCP Server：暴露工具、资源、Prompt 的服务

可以理解为：

Claude Desktop = Host
Claude 内部 MCP 连接模块 = Client
你写的 wiki_mcp_server.py = Server

4.3 MCP 的核心能力

MCP Server 通常可以暴露三类能力：

能力	说明
Tools	可被模型调用的函数，例如查数据库、发请求、计算
Resources	可被读取的资源，例如文件、文档、上下文
Prompts	可复用的提示词模板

你给出的几个核心方法非常典型：

initialize
tools/list
tools/call
resources/list
resources/read
prompts/list

其中最常用的是：

initialize
tools/list
tools/call

4.4 MCP initialize：协议握手

初始化请求示例：

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "initialize",
  "params": {
    "protocolVersion": "2024-11-05",
    "capabilities": {},
    "clientInfo": {
      "name": "claude-desktop",
      "version": "1.0.0"
    }
  }
}

服务端响应：

{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "protocolVersion": "2024-11-05",
    "capabilities": {
      "tools": {},
      "resources": {},
      "prompts": {}
    },
    "serverInfo": {
      "name": "company-wiki-mcp",
      "version": "1.0.0"
    }
  }
}

这个过程类似 LSP 的 initialize。

它的作用是：

• 客户端告诉服务端自己是谁。
• 服务端告诉客户端自己支持什么能力。
• 双方确认协议版本。
• 双方交换 capabilities。

4.5 MCP tools/list：列出可用工具

客户端请求：

{
  "jsonrpc": "2.0",
  "id": 2,
  "method": "tools/list",
  "params": {}
}

服务端返回：

{
  "jsonrpc": "2.0",
  "id": 2,
  "result": {
    "tools": [
      {
        "name": "search_company_wiki",
        "description": "搜索公司内部 Wiki，适合查询制度、流程、项目文档",
        "inputSchema": {
          "type": "object",
          "properties": {
            "query": {
              "type": "string",
              "description": "搜索关键词"
            }
          },
          "required": ["query"]
        }
      }
    ]
  }
}

这个结果非常重要。

因为模型并不是直接“知道”你的工具，而是通过 tools/list 获取工具列表。

模型看到工具描述后，才能判断：

用户问的是公司制度问题，应该调用 search_company_wiki。

4.6 MCP tools/call：执行工具

客户端调用：

{
  "jsonrpc": "2.0",
  "id": 3,
  "method": "tools/call",
  "params": {
    "name": "search_company_wiki",
    "arguments": {
      "query": "年假申请流程"
    }
  }
}

服务端返回：

{
  "jsonrpc": "2.0",
  "id": 3,
  "result": {
    "content": [
      {
        "type": "text",
        "text": "年假申请需要在 OA 系统提交请假单，直属领导审批后生效。"
      }
    ]
  }
}

这就是 MCP 的核心闭环：

发现工具 → 理解工具 → 调用工具 → 返回结果 → 模型组织回答

4.7 MCP 的本质

MCP 不是简单的函数调用。

它真正解决的是：

AI 应用和外部工具之间的标准化连接问题。

以前：

Claude 接 Wiki 要写一套
Cursor 接 Wiki 要写一套
自研 Agent 接 Wiki 又要写一套

现在：

只要实现一个 Wiki MCP Server
多个 MCP Client 都可以接入

所以 MCP 可以理解为：

AI 工具生态里的 USB-C 接口

---

五、ACP：让 IDE 可以唤起和控制 Agent

ACP 全称是 Agent Client Protocol。

它解决的问题和 MCP 不一样。

MCP 是：

AI 调用工具

ACP 是：

IDE 调用 Agent

ACP 官方介绍中明确说，它标准化了代码编辑器和 coding agents 之间的通信；相关文档也说明 ACP 面向 agent-editor 集成，而如果你的目标是让 agent 调用外部工具，则应该看 MCP。(GitHub)

5.1 ACP 的典型场景

用户在 IDE 里说：

帮我重构这个函数。

这时不是简单调用一个工具，而是让一个 Agent 接管任务。

Agent 可能需要：

1. 读取当前文件。
2. 分析项目结构。
3. 修改多个文件。
4. 运行测试。
5. 返回 diff。
6. 请求用户确认。
7. 支持取消任务。
8. 支持中途 steer，比如“慢一点”“更保守一点”“不要改接口”。

这和 MCP 的“调用一个工具”完全不是一个层级。

5.2 ACP 和 LSP 的关系

ACP 很像 LSP。

LSP 是：

IDE <-> Language Server

ACP 是：

IDE <-> Coding Agent

LSP 解决的是：

补全、跳转、诊断、hover

ACP 解决的是：

任务执行、文件修改、进度更新、权限请求、取消、中途引导

所以可以这样理解：

LSP 是 IDE 接入语言智能的协议。
ACP 是 IDE 接入 AI Agent 的协议。

5.3 ACP 的核心流程

根据 ACP 官方概览，典型流程包括：客户端发送 session/prompt，Agent 发送 session/update 通知进度，必要时 Agent 向客户端请求文件操作或权限，客户端可发送 session/cancel 中断任务，最终 session/prompt 返回 stop reason。(agentclientprotocol.com)

一个简化流程：

1. initialize
2. session/new
3. session/prompt
4. session/update
5. session/cancel
6. session/prompt response

不同版本和实现中方法名可能略有差异。例如你提到的 newSession、prompt、cancel、/steer、/queue 更像是概念层或某些实现中的命令表达；当前公开文档中更常见的是 session/prompt、session/update、session/cancel 这类命名。ACP 仍处于发展阶段，部分实现和命名可能会演进。(goose-docs.ai)

5.4 ACP initialize：客户端注册

示例：

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "initialize",
  "params": {
    "protocolVersion": "0.1.0",
    "clientInfo": {
      "name": "cursor",
      "version": "1.0.0"
    },
    "capabilities": {
      "fileSystem": true,
      "terminal": true,
      "diff": true
    }
  }
}

Agent 返回：

{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "agentInfo": {
      "name": "my-coding-agent",
      "version": "1.0.0"
    },
    "capabilities": {
      "streaming": true,
      "fileEdit": true,
      "terminalCommand": true,
      "cancellation": true
    }
  }
}

这里交换的是：

客户端能提供什么上下文？
Agent 能执行什么任务？
双方支持哪些能力？

5.5 ACP 创建会话

一个 coding agent 往往是会话化的。

因为它需要记住：

• 当前任务目标
• 当前项目路径
• 已经读取过的文件
• 已经修改过的文件
• 当前计划
• 用户中途补充的约束

示例：

{
  "jsonrpc": "2.0",
  "id": 2,
  "method": "session/new",
  "params": {
    "workspace": "file:///Users/me/project",
    "mode": "edit"
  }
}

返回：

{
  "jsonrpc": "2.0",
  "id": 2,
  "result": {
    "sessionId": "sess_123"
  }
}

5.6 ACP prompt：发送任务

用户在 IDE 里说：

帮我把这个函数拆成三个小函数，并补充单元测试。

对应请求：

{
  "jsonrpc": "2.0",
  "id": 3,
  "method": "session/prompt",
  "params": {
    "sessionId": "sess_123",
    "prompt": "帮我把这个函数拆成三个小函数，并补充单元测试。"
  }
}

Agent 开始工作。

这时它可能会不断发送进度通知：

{
  "jsonrpc": "2.0",
  "method": "session/update",
  "params": {
    "sessionId": "sess_123",
    "message": "正在分析项目结构..."
  }
}

再比如：

{
  "jsonrpc": "2.0",
  "method": "session/update",
  "params": {
    "sessionId": "sess_123",
    "message": "已找到目标函数，准备修改 service.py 和 test_service.py"
  }
}

最后返回：

{
  "jsonrpc": "2.0",
  "id": 3,
  "result": {
    "stopReason": "completed",
    "summary": "已完成函数拆分，并新增 3 个单元测试。"
  }
}

5.7 ACP cancel：中断任务

Agent 任务可能很长。

用户可能中途发现方向不对，于是点击取消。

{
  "jsonrpc": "2.0",
  "id": 4,
  "method": "session/cancel",
  "params": {
    "sessionId": "sess_123"
  }
}

Agent 收到后应该尽量停止当前任务。

返回：

{
  "jsonrpc": "2.0",
  "id": 4,
  "result": {
    "cancelled": true
  }
}

5.8 steer：中途引导

你提到的 /steer 很关键。

它不是简单的“新任务”，而是对当前任务的方向修正。

例如用户说：

/steer 慢一点，先不要改文件，先给我解释计划。

Agent 应该调整策略：

从直接修改模式 → 计划说明模式

或者：

/steer 保守一点，不要改公共接口。

Agent 应该把这个约束加入当前会话。

这类能力是 coding agent 和普通 chat bot 的重要区别。

5.9 queue：排队消息

当 Agent 正在执行长任务时，用户可能又发来一句：

顺便把测试也补上。

如果立即打断当前任务，可能会导致状态混乱。

所以可以设计 /queue：

当前任务继续执行
新消息进入队列
当前任务完成后再处理

这对于长任务 Agent 非常重要。

---

六、MCP 和 ACP 的核心差异

MCP 和 ACP 很容易混淆。

其实它们解决的问题完全不同。

对比项	MCP	ACP
全称	Model Context Protocol	Agent Client Protocol
主要对象	AI 客户端与外部工具	IDE 与 coding agent
核心问题	让模型发现并调用工具	让 IDE 唤起、控制、协作 Agent
典型客户端	Claude Desktop、Cursor、自研 Agent Host	Cursor、VS Code、Zed 等 IDE
典型服务端	Wiki MCP Server、DB MCP Server、Git MCP Server	Coding Agent、Deep Agent、Goose
通信风格	工具发现与调用	会话、任务、进度、文件操作、权限
类比	AI 时代的工具 USB-C	AI Agent 时代的 LSP
主动/被动	工具被 AI 调用，偏被动	Agent 被 IDE 唤起执行任务，偏主动

可以这样记：

MCP：让 Agent 会“用工具”。
ACP：让 IDE 会“用 Agent”。

或者更直接：

MCP 是 Tool Protocol。
ACP 是 Agent Protocol。

LangChain 文档中也明确区分了两者：ACP 用于 coding agents 和 code editors / IDEs 通信；如果需要 agent 调用外部服务器上的工具，应参考 MCP。(LangChain 文档)

---

七、Gateway Plugin：把 Agent 接入聊天平台

MCP 和 ACP 主要面向 AI 客户端、IDE、工具生态。

但企业里还有另一个常见需求：

我希望用户可以在微信、Slack、Discord、Telegram 里直接和 Agent 对话。

这时候就需要 Gateway Plugin。

它不是一个像 MCP / ACP 那样的统一协议标准，更像是一种插件化架构模式。

你的设计是这样的：

gateway/
  platforms/
    wechat/
    slack/
    discord/
    telegram/
    ...

每个平台一个插件包。

每个插件实现统一接口：

class PlatformPlugin:
    def receive_message(self) -> Message:
        ...

    def send_message(self, msg: Message) -> None:
        ...

    def authenticate(self) -> bool:
        ...

7.1 为什么需要 Gateway Plugin？

因为不同聊天平台的接入方式完全不同。

Slack 使用 Events API 时，Slack 会把订阅的事件发送给你的应用；Slack 文档中也说明 Events API 支持通过 Socket Mode 或指定公开 HTTP endpoint 接收事件。(Slack 开发者文档)

Telegram Bot API 则支持 getUpdates 和 webhook 两种互斥的更新接收方式，收到的是 JSON 序列化的 Update 对象。(core.telegram.org)

不同平台差异包括：

平台	接收消息方式	发送消息方式	特殊能力
Slack	Events API / Socket Mode	Web API	Thread、Channel、App Mention
Telegram	getUpdates / Webhook	Bot API	Chat ID、Inline Keyboard
Discord	Gateway / HTTP API	REST / Gateway	Guild、Channel、Interaction
微信	回调 / 轮询 / 企业微信 API	平台 API	企业身份、群聊、审批流

如果业务层直接适配每个平台，就会非常混乱。

所以需要 Gateway Plugin 做一层抽象。

7.2 Gateway Plugin 的核心目标

Gateway Plugin 解决的是：

不同聊天平台输入输出不统一的问题

它把平台消息统一转换为内部标准消息：

@dataclass
class Message:
    platform: str
    conversation_id: str
    user_id: str
    text: str
    attachments: list
    raw: dict

然后交给 Agent Runtime：

Slack Message
Telegram Update
Discord Event
Wechat Message
        |
        v
Gateway Plugin
        |
        v
统一 Message 对象
        |
        v
Agent Runtime

Agent 返回结果后，再由对应插件转换回平台格式：

Agent Response
        |
        v
Gateway Plugin
        |
        v
Slack / Telegram / Discord / WeChat

7.3 Gateway Plugin 架构示例

+----------------------+
|      Slack Plugin    |
+----------------------+
          |
+----------------------+
|    Telegram Plugin   |
+----------------------+
          |
+----------------------+
|     WeChat Plugin    |
+----------------------+
          |
          v
+----------------------+
|   Gateway Core       |
| - auth               |
| - routing            |
| - session mapping    |
| - rate limit         |
| - retry              |
+----------------------+
          |
          v
+----------------------+
|   Agent Runtime      |
| - memory             |
| - tools              |
| - skills             |
| - model              |
+----------------------+

7.4 Gateway Plugin 的关键模块

1. 平台认证

def authenticate(self) -> bool:
    ...

不同平台认证方式不同：

• Slack：OAuth / Bot Token
• Telegram：Bot Token
• Discord：Bot Token
• 企业微信：CorpID / Secret / AgentID

插件层负责处理平台认证，不要把这些细节暴露给 Agent Runtime。

2. 消息接收

def receive_message(self) -> Message:
    ...

不同平台收到的原始消息结构完全不同。

插件要把它转成统一 Message。

例如 Slack 原始事件：

{
  "type": "message",
  "user": "U123",
  "channel": "C123",
  "text": "帮我总结一下今天的工作"
}

统一成：

{
  "platform": "slack",
  "conversation_id": "C123",
  "user_id": "U123",
  "text": "帮我总结一下今天的工作"
}

3. 消息发送

def send_message(self, msg: Message) -> None:
    ...

Agent 返回的统一结构：

{
  "conversation_id": "C123",
  "text": "今天你主要完成了三个事项……"
}

Slack 插件转换成 Slack Web API 调用。

Telegram 插件转换成 sendMessage。

Discord 插件转换成 Discord 消息发送。

4. 会话映射

同一个用户可能在多个平台出现：

Slack user U123
Telegram user T456
企业微信 user W789

Gateway 需要维护身份映射：

platform_user_id -> internal_user_id

这样才能共享 Agent 记忆和上下文。

5. 平台能力降级

不同平台支持的能力不同。

例如：

• Slack 支持 Thread。
• Telegram 支持 Inline Keyboard。
• Discord 支持 Slash Command。
• 微信可能有自己的卡片消息。

Agent Runtime 不应该关心这些差异。

Gateway Plugin 应该做能力适配：

class PlatformCapabilities:
    supports_thread: bool
    supports_markdown: bool
    supports_file_upload: bool
    supports_buttons: bool

如果平台不支持某种能力，就降级成普通文本。

---

八、把 MCP、ACP、Gateway Plugin 放到一个系统里

在一个完整 AI Agent 系统中，这三者可以同时存在。

例如你要做一个企业 AI 助手：

用户入口：
- IDE
- Claude Desktop
- 企业微信
- Slack

Agent 能力：
- 查询内部 Wiki
- 查询数据库
- 读取代码仓库
- 修改代码
- 生成日报

这时候架构可以是：

                  +-------------------+
                  |   Claude Desktop  |
                  +-------------------+
                            |
                            | MCP
                            v
+-------------------+    +-------------------+
|   Wiki MCP Server |    |   DB MCP Server   |
+-------------------+    +-------------------+


+-------------------+
|   Cursor / IDE    |
+-------------------+
          |
          | ACP
          v
+-------------------+
|   Coding Agent    |
+-------------------+


+-------------------+     +-------------------+
| Slack / WeChat    | --> | Gateway Plugin    |
| Telegram / Discord|     | Platform Adapter  |
+-------------------+     +-------------------+
                                    |
                                    v
                            +-------------------+
                            |   Agent Runtime   |
                            +-------------------+

8.1 分工关系

模块	解决什么问题
MCP	Agent 如何调用外部工具
ACP	IDE 如何驱动 Agent 执行代码任务
Gateway Plugin	聊天平台如何接入 Agent
LSP	IDE 如何获得语言能力
JSON-RPC	请求、响应、通知的消息格式
stdio / HTTP / SSE	消息传输通道

8.2 一个完整调用链示例

用户在 Slack 里问：

公司年假怎么申请？

流程：

1. Slack Plugin 收到 message event
2. Gateway 转成统一 Message
3. Agent Runtime 接收用户问题
4. Agent 判断需要查询内部制度
5. Agent 通过 MCP 调用 wiki_search 工具
6. Wiki MCP Server 返回制度内容
7. Agent 组织回答
8. Gateway 调用 Slack Plugin 发回消息

调用链：

Slack
  -> Gateway Plugin
  -> Agent Runtime
  -> MCP Client
  -> Wiki MCP Server
  -> Agent Runtime
  -> Gateway Plugin
  -> Slack

用户在 Cursor 里说：

帮我重构 order_service.py，并补充测试。

流程：

1. Cursor 通过 ACP 创建 session
2. Cursor 发送 session/prompt
3. Coding Agent 分析项目文件
4. Agent 请求文件读写权限
5. Agent 修改代码
6. Agent 返回进度 update
7. Agent 返回最终 diff 和总结

调用链：

Cursor
  -> ACP
  -> Coding Agent
  -> File System / Tools
  -> ACP Update
  -> Cursor UI

---

九、为什么这些协议都喜欢“initialize + capabilities”？

你会发现 LSP、MCP、ACP 都有一个共同点：

第一步都是 initialize。

这是为什么？

因为这类协议都不是一次性 HTTP API，而是长连接、长生命周期、双向协作。

双方需要先交换能力。

例如：

LSP initialize

编辑器问语言服务器：

你支持 hover 吗？
你支持 completion 吗？
你支持 definition 吗？

语言服务器返回：

我支持 hover、completion、diagnostics。

MCP initialize

AI Client 问 MCP Server：

你支持 tools 吗？
你支持 resources 吗？
你支持 prompts 吗？

MCP Server 返回：

我支持 tools/list 和 tools/call。

ACP initialize

IDE 问 Coding Agent：

你支持文件编辑吗？
你支持任务取消吗？
你支持流式进度吗？

Agent 返回：

我支持 file edit、streaming update、cancel。

这就是 capabilities 协商。

它的好处是：

1. 客户端不用提前写死服务端能力。
2. 服务端可以按需暴露能力。
3. 不同版本之间更容易兼容。
4. 可以做渐进式增强。
5. 可以支持实验性能力。

---

十、协议设计中的几个关键概念

10.1 Request / Response

用于需要结果的调用。

例如：

tools/list
tools/call
session/prompt
textDocument/hover

都有明确返回值。

10.2 Notification

用于不需要响应的事件。

例如：

initialized
progress update
file changed
diagnostics published

适合“我告诉你一件事，但你不用回复我”。

10.3 Capability

表示一方支持什么能力。

例如：

{
  "capabilities": {
    "tools": {},
    "resources": {},
    "prompts": {}
  }
}

10.4 Session

表示一次上下文连续的任务。

ACP 中 session 很重要，因为 coding agent 通常不是一次调用就结束，而是持续多轮工作。

10.5 Transport

Transport 是消息怎么传。

常见有：

stdio
HTTP
SSE
WebSocket
Streamable HTTP

协议定义“消息长什么样”，Transport 定义“消息怎么发”。

---

十一、自己实现一个最小 MCP Server

下面用 Python 写一个非常简化的 MCP 风格 Server。

注意：这是教学示例，不是完整 MCP SDK 实现。

import sys
import json


TOOLS = [
    {
        "name": "search_company_wiki",
        "description": "搜索公司内部 Wiki",
        "inputSchema": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "搜索关键词"
                }
            },
            "required": ["query"]
        }
    }
]


def send_response(request_id, result):
    resp = {
        "jsonrpc": "2.0",
        "id": request_id,
        "result": result
    }
    print(json.dumps(resp, ensure_ascii=False), flush=True)


def send_error(request_id, code, message):
    resp = {
        "jsonrpc": "2.0",
        "id": request_id,
        "error": {
            "code": code,
            "message": message
        }
    }
    print(json.dumps(resp, ensure_ascii=False), flush=True)


def handle_initialize(req):
    send_response(req["id"], {
        "protocolVersion": "2024-11-05",
        "capabilities": {
            "tools": {}
        },
        "serverInfo": {
            "name": "demo-wiki-mcp",
            "version": "1.0.0"
        }
    })


def handle_tools_list(req):
    send_response(req["id"], {
        "tools": TOOLS
    })


def handle_tools_call(req):
    params = req.get("params", {})
    name = params.get("name")
    arguments = params.get("arguments", {})

    if name != "search_company_wiki":
        send_error(req["id"], -32601, f"Unknown tool: {name}")
        return

    query = arguments.get("query", "")

    # 这里应该接真实的 Wiki / RAG / 数据库
    result = f"你查询的是：{query}。这里返回公司内部 Wiki 的模拟结果。"

    send_response(req["id"], {
        "content": [
            {
                "type": "text",
                "text": result
            }
        ]
    })


def main():
    for line in sys.stdin:
        line = line.strip()
        if not line:
            continue

        req = json.loads(line)
        method = req.get("method")

        if method == "initialize":
            handle_initialize(req)
        elif method == "tools/list":
            handle_tools_list(req)
        elif method == "tools/call":
            handle_tools_call(req)
        else:
            send_error(req.get("id"), -32601, f"Method not found: {method}")


if __name__ == "__main__":
    main()

这个最小版本实现了：

initialize
tools/list
tools/call

也就是 MCP 最核心的工具调用闭环。

---

十二、自己实现一个最小 ACP 风格 Agent

下面是一个简化版 ACP 风格 Agent。

import sys
import json
import uuid


sessions = {}


def send_response(request_id, result):
    print(json.dumps({
        "jsonrpc": "2.0",
        "id": request_id,
        "result": result
    }, ensure_ascii=False), flush=True)


def send_notification(method, params):
    print(json.dumps({
        "jsonrpc": "2.0",
        "method": method,
        "params": params
    }, ensure_ascii=False), flush=True)


def handle_initialize(req):
    send_response(req["id"], {
        "agentInfo": {
            "name": "demo-coding-agent",
            "version": "1.0.0"
        },
        "capabilities": {
            "streaming": True,
            "cancellation": True,
            "fileEdit": False
        }
    })


def handle_session_new(req):
    session_id = str(uuid.uuid4())
    sessions[session_id] = {
        "messages": [],
        "cancelled": False
    }

    send_response(req["id"], {
        "sessionId": session_id
    })


def handle_session_prompt(req):
    params = req.get("params", {})
    session_id = params.get("sessionId")
    prompt = params.get("prompt")

    if session_id not in sessions:
        send_response(req["id"], {
            "stopReason": "error",
            "message": "Session not found"
        })
        return

    sessions[session_id]["messages"].append({
        "role": "user",
        "content": prompt
    })

    send_notification("session/update", {
        "sessionId": session_id,
        "message": "正在分析任务..."
    })

    send_notification("session/update", {
        "sessionId": session_id,
        "message": "正在制定执行计划..."
    })

    send_response(req["id"], {
        "stopReason": "completed",
        "summary": f"已收到任务：{prompt}。这里是模拟执行结果。"
    })


def handle_session_cancel(req):
    params = req.get("params", {})
    session_id = params.get("sessionId")

    if session_id in sessions:
        sessions[session_id]["cancelled"] = True

    send_response(req["id"], {
        "cancelled": True
    })


def main():
    for line in sys.stdin:
        line = line.strip()
        if not line:
            continue

        req = json.loads(line)
        method = req.get("method")

        if method == "initialize":
            handle_initialize(req)
        elif method == "session/new":
            handle_session_new(req)
        elif method == "session/prompt":
            handle_session_prompt(req)
        elif method == "session/cancel":
            handle_session_cancel(req)
        else:
            send_response(req.get("id"), {
                "error": f"Unknown method: {method}"
            })


if __name__ == "__main__":
    main()

这个示例体现了 ACP 的核心思想：

不是调用一个工具，而是创建一个 Agent 会话，让 Agent 持续处理任务。

---

十三、Gateway Plugin 最小实现

下面是一个简单的插件抽象。

from dataclasses import dataclass
from typing import Any, Dict, List


@dataclass
class Message:
    platform: str
    conversation_id: str
    user_id: str
    text: str
    attachments: List[Any]
    raw: Dict[str, Any]


class PlatformPlugin:
    name: str

    def authenticate(self) -> bool:
        raise NotImplementedError

    def receive_message(self) -> Message:
        raise NotImplementedError

    def send_message(self, msg: Message) -> None:
        raise NotImplementedError

Slack 插件示例：

class SlackPlugin(PlatformPlugin):
    name = "slack"

    def authenticate(self) -> bool:
        # 校验 bot token
        return True

    def receive_message(self) -> Message:
        # 从 Slack Events API 接收事件
        event = {
            "channel": "C123",
            "user": "U123",
            "text": "帮我总结一下今天的工作"
        }

        return Message(
            platform="slack",
            conversation_id=event["channel"],
            user_id=event["user"],
            text=event["text"],
            attachments=[],
            raw=event
        )

    def send_message(self, msg: Message) -> None:
        # 调用 Slack Web API 发送消息
        print(f"[Slack] send to {msg.conversation_id}: {msg.text}")

Telegram 插件示例：

class TelegramPlugin(PlatformPlugin):
    name = "telegram"

    def authenticate(self) -> bool:
        # 校验 bot token
        return True

    def receive_message(self) -> Message:
        update = {
            "message": {
                "chat": {"id": "T_CHAT_123"},
                "from": {"id": "T_USER_456"},
                "text": "帮我查一下年假制度"
            }
        }

        msg = update["message"]

        return Message(
            platform="telegram",
            conversation_id=str(msg["chat"]["id"]),
            user_id=str(msg["from"]["id"]),
            text=msg["text"],
            attachments=[],
            raw=update
        )

    def send_message(self, msg: Message) -> None:
        # 调用 Telegram sendMessage
        print(f"[Telegram] send to {msg.conversation_id}: {msg.text}")

Gateway Core：

class Gateway:
    def __init__(self):
        self.plugins = {}

    def register(self, plugin: PlatformPlugin):
        if plugin.authenticate():
            self.plugins[plugin.name] = plugin

    def handle_message(self, plugin_name: str):
        plugin = self.plugins[plugin_name]

        incoming = plugin.receive_message()

        response_text = self.call_agent(incoming)

        outgoing = Message(
            platform=incoming.platform,
            conversation_id=incoming.conversation_id,
            user_id=incoming.user_id,
            text=response_text,
            attachments=[],
            raw={}
        )

        plugin.send_message(outgoing)

    def call_agent(self, msg: Message) -> str:
        # 这里可以调用真实 Agent Runtime
        return f"Agent 收到你的消息：{msg.text}"

使用：

gateway = Gateway()
gateway.register(SlackPlugin())
gateway.register(TelegramPlugin())

gateway.handle_message("slack")
gateway.handle_message("telegram")

---

十四、这几类协议应该怎么选？

场景一：我想让 Claude / Cursor 调用我写的工具

用 MCP。

例如：

查数据库
查 Wiki
读文件
调用内部 API
查订单
查日志

你应该写：

xxx_mcp_server.py

暴露：

tools/list
tools/call

场景二：我想让 IDE 调用我的 coding agent

用 ACP。

例如：

帮我重构代码
帮我修 bug
帮我生成测试
帮我解释项目结构
帮我迁移接口

你应该实现：

initialize
session/new
session/prompt
session/update
session/cancel

场景三：我想让 Agent 接入微信、Slack、Telegram

用 Gateway Plugin。

例如：

企业微信 AI 助手
Slack 工作流助手
Telegram 私人助理
Discord 社区机器人

你应该设计：

PlatformPlugin
receive_message
send_message
authenticate

场景四：我想给 IDE 做代码补全、跳转、诊断

用 LSP。

例如：

自研 DSL 的语法检查
SQL 编辑器智能提示
配置文件 schema 校验
工作流 DSL 补全

你应该实现：

initialize
textDocument/didOpen
textDocument/didChange
textDocument/completion
textDocument/hover
textDocument/publishDiagnostics

---

十五、总结

AI Agent 时代，协议会变得越来越重要。

因为我们不再只是做一个“调用大模型的应用”，而是在构建一个复杂生态：

AI Client
IDE
Agent
Tool
Resource
Prompt
File System
Chat Platform
Enterprise System

这些组件之间如果没有统一协议，就会变成大量脆弱的胶水代码。

本文介绍的几个协议和架构，可以这样理解：

JSON-RPC：消息格式底座
stdio / HTTP / SSE：传输方式
LSP：IDE 调语言能力
MCP：AI 调外部工具
ACP：IDE 调 AI Agent
Gateway Plugin：聊天平台接入 Agent

它们之间不是替代关系，而是分工关系。

最终可以总结成一句话：

LSP 解决“编辑器如何使用语言能力”，MCP 解决“模型如何使用外部工具”，ACP 解决“IDE 如何驱动 Agent”，Gateway Plugin 解决“不同聊天平台如何接入 Agent”。

如果你正在做企业级 Agent、AI IDE、内部知识库助手、OA 智能助手、代码智能体平台，这几个协议和架构思想都非常值得深入理解。

因为未来复杂 Agent 系统的竞争，不只是模型能力的竞争，更是：

协议设计能力
工具生态能力
上下文组织能力
多端接入能力
工程化落地能力

谁能把这些能力标准化、模块化、插件化，谁就更容易构建真正可扩展的 AI Agent 平台。