说明

对AI应用开发所涉及到的流程、工具、技能进行系列介绍，全部文章收录于《AI应用开发》专栏。关注+收藏，不错过后续精彩。

前置文章

AI应用开发01-环境准备
AI应用开发02-从零构建AI聊天机器人
AI应用开发03-RAG增强知识问答
AI应用开发04-对话+RAG管理桌面端GUI
AI应用开发05-MCP集成
AI应用开发06-SKILL的诞生
AI应用开发07-深入认识MCP
AI应用开发08-Python前后端分离之FastAPI

项目源码地址

一、项目概述

Model Context Protocol (MCP) 是一种开放标准协议，旨在为 LLM 应用与外部数据源、工具提供统一的集成通道。项目基于 Python 实现了完整的 MCP 客户端与服务端，支持 Resources、Prompts、Tools 三大核心能力，并能够正确处理服务器主动发起的 Elicitation（表单引导）和 Sampling（LLM 采样）请求。所有通信基于 STDIO 传输，遵循 JSON-RPC 2.0 规范。

项目组件：

• MCP 客户端：命令行交互程序，连接服务器子进程，支持列出/调用工具、提示词、资源，并处理服务器发起的反向请求。
• MCP 服务端：提供示例能力（echo、add、code_review 等），可扩展为真实业务服务。

---

二、 整体架构设计

2.1 架构图

2.2 分层解耦的设计原则

层级	职责	关键抽象
协议层	JSON-RPC 消息构造、解析、类型判断	JSONRPC 静态类
生命周期	能力协商、初始化状态管理	ClientLifecycle / LifecycleManager
传输层	STDIO 子进程管理，消息收发	StdioTransport
请求/响应管理	异步回调注册、响应分发	ResponseHandler
服务器请求处理	处理 Elicitation、Sampling 等反向请求	ServerRequestHandler
业务能力层	实现 Resources、Prompts、Tools 具体逻辑	各个 Handler 类
主控层	消息路由、命令解析、流程编排	MCPClient / MCPServer

2.3 完整消息流时序图

初始化握手

工具调用流程

服务器主动 Elicitation (表单)

服务器主动 Sampling

---

三、核心概念与协议规范

3.1 JSON-RPC 2.0 三种消息类型

类型	特征	示例
请求	包含 id, method，期望响应	{"jsonrpc":"2.0","id":1,"method":"tools/list"}
响应	包含 id，必有 result 或 error	{"jsonrpc":"2.0","id":1,"result":{...}}
通知	无 id，无需响应	{"jsonrpc":"2.0","method":"initialized"}

3.2 标准错误码

错误码	名称	说明
-32700	Parse error	JSON 解析错误
-32600	Invalid Request	无效请求对象
-32601	Method not found	方法不存在
-32602	Invalid params	参数无效
-32603	Internal error	内部错误
-32000 ~ -32099	Server error	服务器自定义错误（如用户拒绝采样）

3.3 核心方法说明

方法	调用方向	用途
initialize	C → S	握手，交换能力信息
initialized	C → S (通知)	确认初始化完成
resources/list	C → S	获取资源列表
resources/read	C → S	读取资源内容
prompts/list	C → S	获取提示词列表
prompts/get	C → S	获取填充后的提示词
tools/list	C → S	获取工具列表
tools/call	C → S	调用工具
elicitation/create	S → C	发起用户信息收集表单
sampling/createMessage	S → C	请求 LLM 采样

---

四、 模块详解

4.1 协议层

protocol/jsonrpc.py

• JSONRPC 静态类：提供 create_request, create_response, create_error, create_notification 以及 is_request, is_response, is_notification 判断方法。
• JSONRPCErrorCode：常量定义标准错误码。
• parse_message：安全 JSON 解析函数。

protocol/lifecycle.py（客户端）

• ClientLifecycle：
  • 声明客户端能力 capabilities（roots, sampling, elicitation）。
  • 构建 initialize 参数。
  • 保存服务器返回的 capabilities。

protocol/lifecycle.py（服务端）

• LifecycleManager：
  • 声明服务端能力 server_capabilities（resources, prompts, tools）。
  • 处理 initialize 请求，返回协议版本和能力。
  • 标记初始化状态，防止未授权请求。

4.2 传输层

客户端 transport/stdio_transport.py

• StdioTransport：
  • 使用 subprocess.Popen 启动服务器子进程，重定向 stdin/stdout。
  • start(on_message)：启动读取线程，每收到一行 JSON 就调用回调。
  • send(message)：写入子进程 stdin。
  • 支持优雅终止子进程。

服务端 transport/stdio.py

• 简化版本：仅提供同步的 readline/write 封装，在主循环中使用。

4.3 Handlers

handlers/response_handler.py

• ResponseHandler：
  • 维护 _pending 字典（请求 id → 回调函数）。
  • register(callback)：生成唯一 id，存储回调，返回 id。
  • on_response(id, result)：触发对应回调。
  • 线程安全（使用 threading.Lock）。

handlers/server_requests.py

• ServerRequestHandler：
  • 处理来自服务器的 elicitation/create 和 sampling/createMessage 请求。
  • _handle_elicitation：在终端展示表单，收集用户输入，返回 accept/decline。
  • _handle_sampling：显示对话历史，让用户手动输入模拟 LLM 响应，或调用真实 API（可扩展）。

4.4 Capabilities（服务端）

capabilities/resources.py

• ResourceHandler：管理资源字典，实现 list_resources 和 read_resource。

capabilities/prompts.py

• PromptHandler：管理提示词模板，实现 list_prompts 和 get_prompt（填充参数）。

capabilities/tools.py

• ToolHandler：定义工具列表（echo, add, long_task），实现 call_tool 执行逻辑。

capabilities/elicitation.py（服务端主动发起）

• ElicitationSender：持有 send 函数，提供 request_user_input 方法，向客户端发送 elicitation/create 请求，并注册回调等待响应。

capabilities/sampling.py（服务端主动发起）

• SamplingSender：提供 request_llm 方法，发送 sampling/createMessage 请求，等待客户端返回 LLM 生成结果。

4.5 MCPClient 主控

client.py 核心流程：

1. 解析命令行参数，构造服务器命令（[sys.executable, server_script]）。
2. 初始化 StdioTransport 并启动。
3. 发送 initialize 同步请求，等待响应；发送 initialized 通知。
4. 启动 REPL 循环，解析用户命令。
5. 对于 tools/list, tools/call 等命令，通过 ResponseHandler 注册回调，发送请求，异步等待结果（同步化通过 threading.Event 简化，实际使用回调）。
6. 后台线程持续处理服务器发来的请求（elicitation/sampling），调用 ServerRequestHandler。

4.6 MCPServer 主控

server.py 核心流程：

1. 创建 LifecycleManager, ResourceHandler, PromptHandler, ToolHandler 等。
2. 进入主循环，从 sys.stdin 逐行读取 JSON。
3. 解析消息类型：
   • 请求：若为 initialize 则调用生命周期管理器，否则检查初始化状态后分发给 _dispatch_request。
   • 响应：转发给 ElicitationSender/SamplingSender 的回调处理。
   • 通知：记录日志（如 initialized 完成初始化）。
4. 业务请求路由到对应的 Handler 生成响应，通过 sys.stdout 返回。

---

五、关键流程

5.1 初始化握手（6步）

1. 客户端 start() → 启动子进程。
2. 客户端发送 initialize 请求（id=0），携带客户端 capabilities。
3. 服务端接收，调用 lifecycle.handle_initialize，保存客户端能力，返回服务端 capabilities 和协议版本。
4. 客户端收到响应，设置 lifecycle.initialized = True，保存服务端能力。
5. 客户端发送 initialized 通知。
6. 服务端收到通知，标记已初始化，开始处理业务请求。

5.2 工具调用

# 客户端 REPL 输入：call add {"a":3,"b":5}
# 内部调用 _send_request("tools/call", params, callback)
def _send_request(method, params, callback):
    req_id = resp_handler.register(callback)
    msg = JSONRPC.create_request(method, params, req_id)
    transport.send(msg)

服务端处理：

elif method == "tools/call":
    name = params["name"]
    args = params.get("arguments", {})
    result = tool_handler.call_tool(name, args)
    resp = JSONRPC.create_response(request_id, result)
    sys.stdout.write(resp + "\n")

5.3 服务器主动 Elicitation

服务端在某个工具中调用：

elicitation.request_user_input(
    message="请填写乘车信息",
    schema=seat_schema,
    on_response=lambda content, action: ... # 继续业务逻辑
)

实际会发送 elicitation/create 请求。

客户端 ServerRequestHandler._handle_elicitation 会：

• 打印 message 和字段提示。
• 循环等待用户输入每个字段。
• 构造 accept 响应并发送。

5.4 服务器主动 Sampling

服务端调用：

sampling.request_llm(
    messages=[{"role":"user","content":...}],
    on_response=lambda result: ...
)

客户端处理类似，展示对话历史，等待用户手动输入模拟响应，或调用真实 LLM API 后返回。

---

六、关键要点与最佳实践

6.1 协议设计要点

1. 严格区分消息类型：请求必须有唯一 id；通知不能有 id；响应必须与请求 id 匹配。
2. 初始化顺序不可颠倒：必须完成 initialize → 响应 → initialized 通知后才能发送业务消息。
3. 能力协商：客户端和服务端都应检查对方 capabilities，不应调用未声明的方法。

6.2 传输层注意事项

• 消息边界：STDIO 下每条消息必须是一行（JSON 内部不能有换行符）。使用 json.dumps 确保紧凑。
• 子进程管理：确保 Popen 的 stderr 重定向到 sys.stderr 以便调试；stdin/stdout 使用 text=True 和 bufsize=1（行缓冲）。
• 并发安全：ResponseHandler 使用锁保护 _pending；写入 stdout 时无需加锁，但注意多线程写入顺序（最好使用队列序列化）。

6.3 错误处理策略

• 传输层：捕获 BrokenPipeError，子进程异常退出时清理。
• 协议层：解析 JSON 失败返回 -32700；方法不存在返回 -32601。
• 业务层：工具执行异常返回 isError=true 的 result，而不是 JSON-RPC 错误（保持协议级成功）。

6.4 安全性考虑

• 生产环境切勿将敏感信息（如 API key）通过 args 传递；使用环境变量。
• 对于 Elicitation 的 Form 模式，禁止收集密码、令牌等敏感数据；必须使用 URL 模式。
• STDIO 传输天然安全（无网络暴露），但若改用 HTTP 传输需实现认证和授权。

---

七、 运行与测试

7.1 启动命令

启动服务端（单独测试）：

python server.py

服务端将等待 STDIN 输入，可手动粘贴 JSON 消息测试。

启动客户端并连接服务端：

python client.py /path/to/server.py

如：
 py .\client.py ..\mcp-server\server.py

或使用绝对路径：

python client.py /path/to/server.py

7.2 交互命令参考

命令	示例	说明
tools	tools	列出所有可用工具
call <name> <json>	call add {"a":3,"b":5}	调用工具
prompts	prompts	列出提示词
get_prompt <name> <json>	get_prompt code_review {"code":"..."}	获取提示词内容
resources	resources	列出资源
read <uri>	read file:///data/readme.txt	读取资源
exit	exit	退出客户端

7.3 演示效果

八、 扩展建议

8.1 功能增强

• 真实 LLM 集成：在 ServerRequestHandler._handle_sampling 中调用 OpenAI/Anthropic API，实现自动采样。
• 支持 Streamable HTTP 传输：替换 STDIO 传输为 aiohttp 实现，支持按需流式响应。

8.2 架构优化

• 异步化：使用 asyncio + anyio 重写，支持并发请求和流式 I/O。
• 插件化能力加载：通过配置文件动态加载工具/资源/提示词。
• 多服务器管理：客户端同时连接多个服务器，聚合能力。

8.3 协议演进适配

• MCP 2025-11-25 版本已稳定，未来草案版本可能引入无状态化、server/discover 等特性。建议关注 specification/draft 变化。
• 当前实现严格遵循 2025-11-25，升级时需调整 protocolVersion 和 capabilities 结构。

---

附录：代码仓库结构

mcp/
├── client/
│   ├── client.py
│   ├── protocol/
│   │   ├── __init__.py
│   │   ├── jsonrpc.py
│   │   └── lifecycle.py
│   ├── transport/
│   │   ├── __init__.py
│   │   └── stdio_transport.py
│   └── handlers/
│       ├── __init__.py
│       ├── response_handler.py
│       └── server_requests.py
├── server/
│   ├── server.py
│   ├── protocol/
│   │   ├── __init__.py
│   │   ├── jsonrpc.py
│   │   └── lifecycle_server.py
│   ├── transport/
│   │   ├── __init__.py
│   │   └── stdio_transport.py
│   └── capabilities/
│       ├── __init__.py
│       ├── resources.py
│       ├── prompts.py
│       ├── tools.py
│       ├── elicitation.py
│       └── sampling.py
└── README.md