1. 什么是 Streamable HTTP? #
前面用的都是 stdio 传输:客户端启动服务器子进程,通过标准输入/输出传 JSON-RPC 消息。
Streamable HTTP 是另一种标准传输方式:服务器作为独立 HTTP 服务运行,客户端通过 HTTP POST/GET 收发消息,必要时用 SSE(Server-Sent Events) 做流式返回。
| 对比 | stdio | Streamable HTTP |
|---|---|---|
| 服务器怎么跑 | 由客户端当子进程拉起 | 你先手动启动,一直监听端口 |
| 连谁 | 进程的 stdin/stdout | URL,如 http://127.0.0.1:8000/mcp |
| 适合场景 | 本机开发、编辑器插件 | 远程服务、多客户端、已有 Web 服务 |
协议层不变:无论是 stdio 还是 HTTP,上面跑的仍是同一套 JSON-RPC(initialize、tools/list、tools/call 等)。换的是「怎么传」,不是「传什么」。
2. 工作流程 #
sequenceDiagram
participant U as 你
participant S as HTTP 服务器
participant C as 客户端/Inspector
U->>S: ① 启动服务器(监听 8000 端口)
C->>S: ② HTTP 连接 /mcp,initialize 握手
S-->>C: ③ 返回能力与 Session ID
C->>S: ④ tools/list、tools/call 等
S-->>C: ⑤ JSON 或 SSE 流式响应
和 stdio 的关键区别:
| 步骤 | stdio | Streamable HTTP |
|---|---|---|
| 启动服务器 | 客户端自动拉起子进程 | 你先在终端里启动 |
| 连接地址 | 无 URL | http://主机:端口/路径 |
| 多客户端 | 一个子进程对应一个客户端 | 同一 URL 可接多个客户端 |
| 会话 | 进程活着就是会话 | 服务器分配 Session ID(SDK 自动处理) |
下面要关心的默认值:
| 配置项 | FastMCP 默认值 |
|---|---|
| 主机 | 127.0.0.1(仅本机可访问) |
| 端口 | 8000 |
| 路径 | /mcp |
| 完整地址 | http://127.0.0.1:8000/mcp |
3. 启动 HTTP 服务器 #
3.1 最简写法 #
只需把 mcp.run() 的传输参数改成 streamable-http:
# 导入 FastMCP 类
from mcp.server.fastmcp import FastMCP
# 创建一个 FastMCP 服务器实例,名称为 "http-demo"
mcp = FastMCP(name="http-demo")
# 将下面的函数注册为工具,可通过 HTTP 被调用
@mcp.tool()
# 定义一个加法工具函数,用于测试 HTTP 模式的调用功能
def add(a: int, b: int) -> str:
# 返回两个参数相加后的字符串结果
return str(a + b)
# 程序主入口,使用 streamable-http 作为数据传输方式(不再使用 stdin/stdout)
if __name__ == "__main__":
# 启动 FastMCP 服务器,使用 streamable-http 传输
mcp.run(transport="streamable-http")3.2 运行 #
uv run python http_server.py期望输出(节选):
INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)看到上述字样表示服务器已在后台监听。不要关这个终端,去另一个终端连客户端。
3.3 自定义地址和端口 #
需要改 host、port 或路径时,在创建 FastMCP 时传入:
# 创建 FastMCP 服务器实例,设置名称为 "http-server"
mcp = FastMCP(
# 指定服务器名称
name="http-server",
# 设置监听的主机地址,这里建议是 127.0.0.1,避免服务暴露到局域网
host="127.0.0.1",
# 设置 HTTP 服务器端口为 9000
port=9000,
# 设置 HTTP 接口路径为 /mcp
streamable_http_path="/mcp",
)
# 判断是否为主程序入口
if __name__ == "__main__":
# 启动 FastMCP 服务器,使用 streamable-http 作为数据传输方式
mcp.run(transport="streamable-http")改端口或路径后,客户端 URL 要一起改,例如 http://127.0.0.1:9000/mcp。
3.4 和 stdio 服务器的代码差异 #
| 部分 | stdio 服务器 | HTTP 服务器 |
|---|---|---|
| `@mcp.tool()` 等注册 | 完全相同 | 完全相同 |
mcp.run() |
默认或 transport="stdio" |
transport="streamable-http" |
| 业务逻辑 | 完全相同 | 完全相同 |
工具、资源、提示的写法不用改,只改启动方式。
4. 编写 HTTP 客户端 #
4.1 连接方式对比 #
| 传输 | 连接函数 | 来源 |
|---|---|---|
| stdio | stdio_client(server_params) |
mcp.client.stdio |
| HTTP | streamable_http_client(url) |
mcp.client.streamable_http |
连上之后的代码完全一样:ClientSession → initialize() → list_tools() → call_tool()。
4.2 示例 #
# 导入异步IO库 asyncio
import asyncio
# 从官方 SDK 导入客户端会话类 ClientSession
from mcp import ClientSession
# 从官方 SDK 导入基于 Streamable HTTP 的上下文管理器
from mcp.client.streamable_http import streamable_http_client
# 设置 HTTP 服务器地址(需与服务器实际 host/port/path 对应)
SERVER_URL = "http://127.0.0.1:8000/mcp"
# 定义异步主函数
async def main() -> None:
# 使用 streamable_http_client 创建 (read, write, get_session_id) 三元组的异步上下文
# get_session_id 一般用不到,这里用 _ 忽略
async with streamable_http_client(SERVER_URL) as (read, write, _):
# 在异步上下文中创建客户端会话 session
async with ClientSession(read, write) as session:
# 第一步:异步初始化,与服务器进行握手
await session.initialize()
# 第二步:异步获取工具列表
tools = await session.list_tools()
# 打印工具名称列表
print("工具列表:", [t.name for t in tools.tools])
# 第三步:异步调用 add 工具,传入参数 a=3, b=5
result = await session.call_tool("add", {"a": 3, "b": 5})
# 遍历结果内容中的每个 block
for block in result.content:
# 如果 block 有 text 属性,则打印返回结果
if hasattr(block, "text"):
print("调用结果:", block.text)
# 判断当前模块是否被直接运行
if __name__ == "__main__":
# 运行异步主函数
asyncio.run(main()) 4.3 运行(两个终端) #
终端 1 — 保持服务器运行:
uv run http_server.py终端 2 — 跑客户端:
uv run http_client.py期望输出:
工具列表: ['add']
调用结果: 84.4 和 stdio 客户端的差异 #
| 部分 | stdio 客户端 | HTTP 客户端 |
|---|---|---|
| 启动服务器 | stdio_client 自动拉起子进程 |
不会启动服务器,只连已有 URL |
| 连接参数 | StdioServerParameters(command, args) |
URL 字符串 |
initialize 之后 |
相同 | 相同 |
4.5 带自定义请求头(可选) #
远程服务常需在请求里带 API Key 或 Token。用 httpx 客户端传入即可:
# 导入 httpx 库,用于发送 HTTP 请求
import httpx
# 从 mcp.client.streamable_http 模块导入 streamable_http_client,用于创建可流式传输的 HTTP 客户端
from mcp.client.streamable_http import streamable_http_client
# 定义请求头,这里设置了授权信息(Bearer Token)
headers = {"Authorization": "Bearer your-token"}
# 异步上下文中创建一个 httpx 的异步客户端,并设置自定义请求头
async with httpx.AsyncClient(headers=headers) as http_client:
# 异步上下文中创建一个 streamable_http_client 实例,指定服务器的 URL,以及上面创建的 http_client
async with streamable_http_client(
"https://example.com/mcp",
http_client=http_client,
) as (read, write, _):
# 后续的代码与前面的示例相同
...
本地 127.0.0.1 开发一般不需要请求头。
5. 用 Inspector 测试 #
HTTP 模式下 Inspector 不会像 stdio 那样帮你启动子进程对话,需要先手动启动服务器。
5.1 步骤 #
| 步骤 | 操作 |
|---|---|
| 1 | 终端 1 启动 03_http_server.py,确认监听 8000 |
| 2 | 终端 2 用下面命令打开 Inspector |
| 3 | 浏览器里确认 Connected |
| 4 | 在 Tools 里测试 add 工具 |
npx @modelcontextprotocol/inspector --transport http --server-url http://127.0.0.1:8000/mcp5.2 如何判断成功 #
| 信号 | 说明 |
|---|---|
| 连接状态 Connected | HTTP 握手与 initialize 完成 |
Tools 里有 add |
工具注册正常 |
执行 add(3, 5) 返回 8 |
端到端通畅 |
5s.3 常见问题 #
| 现象 | 可能原因 |
|---|---|
| 连接失败 / 超时 | 服务器没启动,或 URL 端口/路径写错 |
| 404 | 路径不是 /mcp,检查 streamable_http_path |
| 工具列表为空 | 服务器脚本有语法错误,看终端 1 的报错 |