火花语音盒(崩铁火花
25.0MB · 2026-03-26
在上一篇基础篇中，我们解决了“为什么要用 MCP”的问题。
这一篇，我们继续回答“MCP 到底是怎么通信的”。
如果你希望 AI 不只是聊天，而是真的能稳定地调用数据库、业务系统或第三方 API，那么协议层就是必须吃透的一环。
MCP 的核心在于两大基础协议：
本文你将获得三件事：
在 MCP 中规定了唯一的标准消息格式，就是 JSON-RPC 2.0。
JSON-RPC 2.0 是一种轻量级的、用于远程过程调用（RPC）的消息交换协议，使用 JSON 作为数据格式。
这种消息协议的好处包括：
在 MCP Client 与 Server 建立连接后，必须进行初始化握手（Handshake）才能开始正常通信。这是 MCP 协议的关键步骤。
初始化流程：
Client 发送初始化请求
initialize 方法发送请求，包含 Client 的 capabilities（能力声明）Server 返回初始化响应
能力协商
初始化完成
list_tools()、list_resources() 等方法初始化时序图：
Client Server
| |
|---- initialize(capabilities, protocolVersion) --->|
| |
|<--- result(serverInfo, capabilities, version) ----|
| |
|---- initialized / 后续 list_tools、call_tool ---->|
| |
代码示例：
# Client 端初始化
async with stdio_client(server_params) as (read_stream, write_stream):
session = ClientSession(read_stream, write_stream)
async with session:
# 必须首先调用 initialize()
capabilities = await session.initialize()
# 查看 Server 支持的能力
print(f"Server capabilities: {capabilities.capabilities}")
# 输出示例：
# {
# "tools": {},
# "resources": {},
# "prompts": {}
# }
# 初始化完成后，才能调用其他方法
tools = await session.list_tools()
resources = await session.list_resources()
初始化的重要性：
如果你第一次接触 MCP，可以先记住这个最小请求结构（以 initialize 为例）：
{
"jsonrpc": "2.0",
"id": 1,
"method": "initialize",
"params": {
"protocolVersion": "2025-03-26",
"capabilities": {
"tools": {},
"resources": {},
"prompts": {}
},
"clientInfo": {
"name": "demo-client",
"version": "1.0.0"
}
}
}
你可以把它理解成一句话：“我是谁、我支持什么、我想按哪个协议版本和你通信。”
MCP 支持三种传输协议，分别适用于不同的使用场景：
STDIO（Standard Input/Output） 是一种基于标准输入（stdin）和标准输出（stdout）的本地通信方式。
MCP Client 启动一个子进程（MCP Server）并通过 stdin 和 stdout 交换 JSON-RPC 消息来实现通信。
基本通信过程：
详细描述：
启动子进程（MCP Server）
消息交换
生命周期管理
适用场景：本地开发、单机部署、简单集成场景。
最小代码（STDIO Client）：
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
import asyncio
server_params = StdioServerParameters(
command="python",
args=["mysqlMCPServer.py"],
)
async def main():
async with stdio_client(server_params) as (read_stream, write_stream):
async with ClientSession(read_stream, write_stream) as session:
await session.initialize()
tools = await session.list_tools()
print(tools)
if __name__ == "__main__":
asyncio.run(main())
SSE（Server-Sent Events，服务器发送事件） 是一种基于 HTTP 协议的单向通信技术，允许 Server 主动实时向 Client 推送消息，Client 只需建立一次连接即可持续接收消息。
特点：
由于 SSE 是一种单向通信的模式，所以它需要配合 HTTP Post 来实现 Client 与 Server 的双向通信。
严格地说，这是一种 HTTP Post（Client → Server）+ HTTP SSE（Server → Client） 的伪双工通信模式。
这种传输模式下：
基本通信过程：
详细描述：
连接建立：Client 首先请求建立 SSE 连接，Server"同意"，然后生成并推送唯一的 Session ID
请求发送：Client 通过 HTTP POST 发送 JSON-RPC 2.0 请求（请求中会带有 Session ID 和 Request ID 信息）
请求接收确认：Server 接收请求后立即返回 202（Accepted）状态码，表示已接受请求
异步处理：Server 应用框架会自动处理请求，根据请求中的参数，决定调用某个工具或资源
结果推送：处理完成后，Server 通过 SSE 通道推送 JSON-RPC 2.0 响应，其中带有对应的 Request ID
结果匹配：Client 的 SSE 连接侦听接收到数据流后，会根据 Request ID 将接收到的响应与之前的请求匹配
重复处理：循环步骤 2-6 这个过程（这里面包含一个 MCP 的初始化过程）
连接断开：在 Client 完成所有请求后，可以选择断开 SSE 连接，会话结束
适用场景：远程部署、实时推送需求。
最小代码（SSE Server）：
from fastapi import FastAPI, Request
from mcp.server.sse import SseServerTransport
from starlette.routing import Mount
from mysqlMCPServer import mcp
app = FastAPI()
sse = SseServerTransport("/messages/")
app.router.routes.append(Mount("/messages", app=sse.handle_post_message))
@app.get("/sse")
async def handle_sse(request: Request):
async with sse.connect_sse(request.scope, request.receive, request._send) as (read_stream, write_stream):
await mcp.run(read_stream, write_stream, mcp.create_initialization_options())
在 MCP 新标准（2025-03-26 版）中，MCP 引入了新的 Streamable HTTP 远程传输机制来代替之前的 HTTP+SSE 的远程传输模式，STDIO 的本地模式不变。
该新标准还在 OAuth 2.1 的授权框架、JSON-RPC 批处理、增强工具注解等方面进行增加和调整，且在 2025 年 5 月 8 日发布的 MCP SDK 1.8.0 版本中正式支持了 Streamable HTTP。
HTTP+SSE 方式存在的问题：
基本通信过程：
主要变化：
/mcp 或 /messages）用于通信Streamable HTTP 的优势：
适用场景：生产环境、云部署、需要高可扩展性的场景。
最小代码（Streamable HTTP Server）：
from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
from starlette.applications import Starlette
from starlette.routing import Mount
from mysqlMCPServer import mcp
session_manager = StreamableHTTPSessionManager(
app=mcp,
event_store=None,
json_response=None,
stateless=True,
)
async def handle_streamable_http(scope, receive, send):
await session_manager.handle_request(scope, receive, send)
starlette_app = Starlette(routes=[Mount("/mcp", app=handle_streamable_http)])
三种模式对比总结：
| 特性 | STDIO | SSE Remote | Streamable HTTP |
|---|---|---|---|
| 连接方式 | 本地子进程 | HTTP + SSE 双通道 | 统一 HTTP 端点 |
| 部署场景 | 本地/单机 | 远程（需长连接） | 远程（支持无状态） |
| 连接可靠性 | 高 | 中等（SSE 断开需重建） | 高（支持自动恢复） |
| 扩展性 | 低 | 中等 | 高（无状态设计） |
| 适用场景 | 开发/简单部署 | 实时推送需求 | 生产环境/云部署 |
通过本文，我们深入了解了 MCP 的两大基础协议：
如果你现在要做技术选型，可以先按这个原则：
下期预告：在下一篇文章中，我们将进入实战环节，包括：
