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']
调用结果: 8

4.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/mcp

5.2 如何判断成功 #

信号 说明
连接状态 Connected HTTP 握手与 initialize 完成
Tools 里有 add 工具注册正常
执行 add(3, 5) 返回 8 端到端通畅

5s.3 常见问题 #

现象 可能原因
连接失败 / 超时 服务器没启动,或 URL 端口/路径写错
404 路径不是 /mcp,检查 streamable_http_path
工具列表为空 服务器脚本有语法错误,看终端 1 的报错