首页 › tutorials › 造远程MCP：流式HTTP上云

造远程MCP：用 Streamable HTTP 把工具上云

概述

前面几轮我们造的 MCP 服务器几乎都是 stdio 本地进程——Claude Code 每次启动一个子进程，通过标准输入输出和它对话。这种方式简单、私密、零部署，但有一个硬伤：工具只活在你的机器上。同事要用？复制代码装依赖。换台电脑？重新配一遍。想给团队共用一个内部数据源？完全做不到。

本教程解决的就是这个问题：把 MCP 服务器从「本地子进程」升级为「网络服务」，用 2025 年规范钦定的 Streamable HTTP 传输上云，加一层 Bearer Token 鉴权，让团队里所有 Claude Code（甚至 Cursor、claude.ai）都能通过一个 URL 共享同一套工具。

适合人群：已经写过至少一个 stdio MCP 服务器、想让工具被多人或多个客户端复用、有基本的 HTTP/Python 概念的开发者。学完你将能独立设计、实现、部署、接入并调试一个生产可用的远程 MCP 服务器。

前置条件

第一步：先想清楚——什么时候该上云

stdio 和 Streamable HTTP 不是「新换旧」，而是两种不同的部署形态，选错形态会带来一堆没必要的复杂度。

维度 stdio（本地进程） Streamable HTTP（远程服务）
谁启动它 Claude Code 拉起子进程 服务端常驻，谁都能连
适用场景 个人、本地、读你本机文件 团队共享、云数据源、需要鉴权
鉴权 不需要（进程隔离） 必须（暴露在网络上）
多客户端 1 对 1 N 对 1，支持水平扩展
状态管理 进程内随便存 要考虑无状态化 / 会话亲和
部署成本 零 要运维、要域名、要 TLS

经验法则：如果工具只用一次、只读本机、只你一个人用，就别上云，stdio 永远是首选。一旦出现「团队共享」「数据源在云上」「想被多个 AI 客户端复用」中任意一条，远程 MCP 就值回票价。

本教程的实战案例是 「内网服务健康巡检」远程 MCP：团队里每个人都能问 Claude「现在哪些服务挂了」，而巡检逻辑、服务清单、鉴权密钥只维护在服务端一处。

第二步：为什么是 Streamable HTTP，而不是 SSE

2025-06-18 规范把旧的 HTTP+SSE 传输废弃了，统一改为 Streamable HTTP。区别如下，理解了才不会在调试时踩坑：

一句话：单一端点、按需流式、可恢复。对开发者来说，你只要暴露一个 /mcp，剩下的框架替你处理。

sequenceDiagram
    participant CC as Claude Code
    participant SRV as 远程 MCP 服务 /mcp
    CC->>SRV: POST initialize (JSON-RPC)
    SRV-->>CC: 200 + initializeResult
    Note over CC,SRV: 握手完成，建立 session
    CC->>SRV: POST tools/call (长任务)
    SRV-->>CC: 200, 升级为 SSE 流
    Note over SRV: 边算边推 notifications/progress
    SRV-->>CC: event: message (最终结果)
    Note right of CC: 网络断开？
    CC->>SRV: POST 携带 Last-Event-ID 头
    SRV-->>CC: 从断点续推剩余 event

第三步：搭项目骨架，用 FastMCP v3

直接手撸 Streamable HTTP 协议太重，用官方 Python SDK 之上的 FastMCP（现归 Prefect 维护，v3 已是主流）。它把 run(transport="http") 一行搞定，还能一键转成 ASGI app 接入任意生产服务器。

mkdir remote-health-mcp && cd remote-health-mcp
uv init --python 3.11
uv add "fastmcp>=2.10" httpx

如果用传统 venv：

python -m venv .venv && source .venv/bin/activate
pip install "fastmcp>=2.10" httpx

最小可跑的服务器 server.py：

# server.py —— 远程 MCP 最小骨架
from fastmcp import FastMCP

mcp = FastMCP("health-mcp")

@mcp.tool
def ping() -> str:
    """连通性自检，返回固定字符串。"""
    return "pong"

if __name__ == "__main__":
    # transport="http" 即 Streamable HTTP，默认监听 /mcp 端点
    mcp.run(transport="http", host="0.0.0.0", port=8000)

运行：python server.py，控制台会打印 Uvicorn running on http://0.0.0.0:8000。此时服务端点就是 http://localhost:8000/mcp。

预期结果：浏览器直接访问 http://localhost:8000/mcp 会返回一个提示（因为 MCP 走的是 JSON-RPC over POST，不接受裸 GET），这是正常的——说明端点活着。

常见问题： - 端口被占：换 port=8001。 - mcp.run 报找不到 transport 参数：FastMCP 版本太旧，升级到 >=2.10。 - 云主机访问不到：确认安全组/防火墙放行 8000，且 host 是 0.0.0.0 而非 127.0.0.1。

第四步：写业务工具，并补上 tool annotations

MCP 工具的 2025 新特性是 tool annotations（工具注解）——一组给客户端看的「行为提示」，让 Claude Code 知道某个工具是否只读、是否有破坏性、是否幂等。远程工具尤其要写准，因为客户端会据此决定要不要二次确认。

# tools.py —— 内网服务健康巡检工具集
import os
import httpx
from fastmcp import FastMCP

mcp = FastMCP("health-mcp")

# 服务清单：真实项目可从环境变量或配置中心读取
SERVICES = {
    "user-api":  "http://user-api.internal:8080/health",
    "order-api": "http://order-api.internal:8081/health",
    "billing":   "http://billing.internal:8082/health",
}
TIMEOUT = 3.0

@mcp.tool(annotations={
    "title": "列出受巡检服务",
    "readOnlyHint": True,      # 只读，无副作用
    "openWorldHint": False,    # 不接触外部开放网络
})
def list_services() -> list[str]:
    """返回当前纳入巡检的服务名清单。"""
    return list(SERVICES.keys())

@mcp.tool(annotations={
    "title": "探测单个服务健康",
    "readOnlyHint": True,
    "idempotentHint": True,    # 重复调用结果一致
})
def check_health(service_name: str) -> dict:
    """
    对指定服务发起健康探测。
    返回 {service, status, latency_ms, detail}。
    status 为 ok / down / unknown。
    """
    url = SERVICES.get(service_name)
    if not url:
        return {"service": service_name, "status": "unknown",
                "latency_ms": 0, "detail": "不在巡检清单内"}
    try:
        with httpx.Client(timeout=TIMEOUT) as client:
            r = client.get(url)
        ms = int(r.elapsed.total_seconds() * 1000)
        ok = r.status_code == 200
        return {"service": service_name, "status": "ok" if ok else "down",
                "latency_ms": ms, "detail": r.text[:200]}
    except Exception as e:
        return {"service": service_name, "status": "down",
                "latency_ms": 0, "detail": str(e)[:200]}

@mcp.tool(annotations={
    "title": "生成全量巡检报告",
    "readOnlyHint": True,
})
def full_report() -> str:
    """对所有服务做一次健康探测，输出 Markdown 报告。"""
    lines = ["# 巡检报告", ""]
    for name in SERVICES:
        res = check_health(name)
        emoji = "🟢" if res["status"] == "ok" else "🔴"
        lines.append(f"- {emoji} **{name}** — {res['status']} "
                     f"({res['latency_ms']}ms): {res['detail']}")
    return "\n".join(lines)

annotation 五个字段速查： - title：给人看的显示名 - readOnlyHint：只读，不改世界 - destructiveHint：可能有破坏性（删数据、发邮件等）——远程工具默认建议显式声明 - idempotentHint：幂等，重复调用等价 - openWorldHint：是否接触「开放世界」（外部 LLM 服务、公网不可信来源）

这些是提示不是强制约束，但客户端会用。比如 Claude Code 对 destructiveHint=true 的工具可能要求确认。声明诚实的注解，是远程 MCP 作者的基本素养。

第五步：暴露成 ASGI 应用，挂上 Bearer Token 鉴权

mcp.run() 够原型用，但生产环境我们要中间件（鉴权、日志、CORS、限流）。FastMCP 提供 mcp.http_app() 把服务器转成标准 ASGI app，然后用 uvicorn 跑，并自由叠加 Starlette 中间件。

鉴权是远程 MCP 的命门。一个完全裸奔的 /mcp 等于把你内网的健康检查能力公开给整个互联网。最小可用方案：共享 Bearer Token——只有携带正确 token 的请求才放行。

# app.py —— ASGI app + Bearer 鉴权中间件
import os
import hmac
from starlette.requests import Request
from starlette.responses import JSONResponse
from starlette.middleware import Middleware
from starlette.middleware.base import BaseHTTPMiddleware

from tools import mcp  # 复用上一步的服务器实例

EXPECTED_TOKEN = os.environ["MCP_BEARER_TOKEN"]  # 从环境变量读，绝不硬编码

class BearerAuth(BaseHTTPMiddleware):
    """校验 Authorization: Bearer <token>，不匹配则 401。"""
    async def dispatch(self, request: Request, call_next):
        auth = request.headers.get("authorization", "")
        token = auth.removeprefix("Bearer ").strip()
        # 用 hmac.compare_digest 防 timing attack
        if not hmac.compare_digest(token, EXPECTED_TOKEN):
            return JSONResponse(
                {"error": "unauthorized"}, status_code=401)
        return await call_next(request)

# 关键：把 MCP 转成 ASGI app，并挂载鉴权中间件
app = mcp.http_app(middleware=[Middleware(BearerAuth)])

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

启动前注入密钥：

export MCP_BEARER_TOKEN="$(openssl rand -hex 32)"
python app.py

预期结果：不带 token 直接 POST /mcp → 401 {"error":"unauthorized"}；带正确 Authorization 头 → 正常走 MCP 协议。

常见问题： - 密钥放哪：环境变量、配置中心或密钥管理服务（Vault）。绝不写进代码、绝不进 git。 - 为什么用 hmac.compare_digest：普通 == 比较会因为「比对到不同字符就返回」泄露 token 长度信息，存在 timing attack。 - 想要更细的鉴权：FastMCP 内置 auth= 参数支持完整 OAuth 2.1 流程（授权码、PKCE、动态客户端注册），适合给外部第三方用户用。共享 Bearer 适合团队内部这种「我信任持有 token 的人」的场景。

第六步：接入 Claude Code

服务跑起来后，用 claude mcp add 把它注册进去。注意 --transport http 就是 Streamable HTTP：

# 本机测试
claude mcp add --transport http health \
  http://localhost:8000/mcp \
  --header "Authorization: Bearer $MCP_BEARER_TOKEN"

上云后换成真实域名（务必走 HTTPS，见第九步）：

claude mcp add --transport http health \
  https://mcp.internal.dev/mcp \
  --header "Authorization: Bearer $MCP_BEARER_TOKEN"

如果偏好把配置写进项目（团队共享、进版本库但不含密钥），用 .mcp.json：

{
  "mcpServers": {
    "health": {
      "type": "streamable-http",
      "url": "https://mcp.internal.dev/mcp",
      "headers": {
        "Authorization": "Bearer ${MCP_BEARER_TOKEN}"
      }
    }
  }
}

.mcp.json 支持 ${ENV_VAR} 展开——团队每人把 token 放在自己环境里，配置文件本身可以安全提交。

验证接入：

claude mcp list          # 看到 health 服务器状态 connected
claude mcp get health    # 看到它的工具列表

然后在 Claude Code 会话里直接问：「帮我跑一次全量巡检报告」，Claude 会自动调用 full_report()。

第七步：用 MCP Inspector 抓包调试

远程 MCP 出问题时，肉眼看不到 JSON-RPC 往返。用官方 MCP Inspector 把每条消息打出来：

# 指定 streamable-http 类型 + 鉴权头
npx @modelcontextprotocol/inspector \
  --transport http \
  --url http://localhost:8000/mcp \
  --header "Authorization: Bearer $MCP_BEARER_TOKEN"

浏览器打开 Inspector 后，你能： 1. 看 initialize 握手回了什么（协议版本、capabilities、服务器信息） 2. 手动点 tools/list、tools/call，看请求/响应原文 3. 观察 SSE 流是否正常升级、有没有 Last-Event-ID 续传

FastMCP 也自带开发者模式：fastmcp dev app.py 会启动一个带可视化面板的 Inspector 集成。

调试清单： - 401 → token 不对或未注入环境变量 - 404 → 端点路径不对（默认是 /mcp，注意有没有尾斜杠） - 握手失败协议版本不匹配 → FastMCP/Claude Code 版本太旧，升级 - 工具调用了但报 schema 错误 → 参数类型注解没写清楚，FastMCP 靠函数签名生成 schema，service_name: str 必须带类型

第八步：上云部署

原型在 localhost，生产要解决三件事：进程常驻、TLS、可扩展。

方案 A：自有 VPS + systemd + nginx（最稳）

/etc/systemd/system/health-mcp.service：

[Unit]
Description=Remote Health MCP Server
After=network.target

[Service]
User=mcp
WorkingDirectory=/opt/remote-health-mcp
EnvironmentFile=/opt/remote-health-mcp/.env
ExecStart=/opt/remote-health-mcp/.venv/bin/uvicorn app:app \
          --host 127.0.0.1 --port 8000 --workers 2
Restart=always

[Install]
WantedBy=multi-user.target

注意 host=127.0.0.0 改成 127.0.0.1——只让本机 nginx 反代，不直接暴露 uvicorn。.env 里放 MCP_BEARER_TOKEN。

nginx 反代 + TLS（用 certbot 自动签 Let's Encrypt）：

server {
    listen 443 ssl http2;
    server_name mcp.internal.dev;

    ssl_certificate     /etc/letsencrypt/live/mcp.internal.dev/fullchain.pem;
    ssl_certificate_key /etc/letsencrypt/live/mcp.internal.dev/privkey.pem;

    location /mcp {
        proxy_pass http://127.0.0.1:8000/mcp;
        proxy_http_version 1.1;
        proxy_set_header Host $host;
        # SSE 关键：关闭缓冲、支持长连接
        proxy_buffering off;
        proxy_read_timeout 86400s;
    }
}

proxy_buffering off 是 SSE 流式响应能正常推送的关键，否则 nginx 会把流攒成一坨一次性返回，客户端体验就是「一直转圈」。

方案 B：Serverless 平台（最快）

方案 C：水平扩展要开 Stateless 模式

多 worker / 多实例时，MCP 默认的有状态 session 会让「请求落到 A 实例、后续消息落到 B 实例」时失败。FastMCP 提供 stateless_http=True，让每个请求自包含、不依赖服务器内存态，配合 nginx 的轮询负载均衡即可水平扩。

mcp = FastMCP("health-mcp", stateless_http=True)

代价：无法依赖进程内状态（比如跨调用的缓存、订阅），需要把状态外置到 Redis 之类。对纯只读的巡检工具，无状态是最省心的选择。

第九步：架构与生命周期总览

flowchart LR
    subgraph 客户端
        A1[Claude Code PC1]
        A2[Claude Code PC2]
        A3[claude.ai / Cursor]
    end
    A1 & A2 & A3 -->|"HTTPS POST /mcp\nAuthorization: Bearer"| NGINX[nginx + TLS]
    NGINX --> LB{负载均衡}
    LB --> W1[uvicorn worker 1]
    LB --> W2[uvicorn worker 2]
    W1 & W2 --> AUTH[BearerAuth 中间件]
    AUTH --> MCP[FastMCP ASGI app]
    MCP --> T1[list_services]
    MCP --> T2[check_health]
    MCP --> T3[full_report]
    T2 --> INT[(内网服务\n/health 端点)]
    style NGINX fill:#e8f5e9
    style MCP fill:#fff3e0

整个系统的核心价值：鉴权、TLS、巡检逻辑、服务清单全部集中在服务端一处，所有客户端只需一个 URL + 一个 token，团队运维成本从「N 人 × M 台机器」降到「1 个服务」。

实战案例：把上面的代码跑通

完整目录：

remote-health-mcp/
├── app.py            # ASGI app + 鉴权中间件（第五步）
├── tools.py          # 工具定义 + annotations（第四步）
├── .env              # MCP_BEARER_TOKEN=...（不提交）
├── .gitignore        # 忽略 .env / .venv
├── .mcp.json         # 客户端配置模板（用 ${} 展开）
└── pyproject.toml

端到端验证：

# 1. 起服务
export MCP_BEARER_TOKEN="$(openssl rand -hex 32)"
echo "$MCP_BEARER_TOKEN" > .env  # 仅本机，勿提交
python app.py &

# 2. 接入 Claude Code
claude mcp add --transport http health http://localhost:8000/mcp \
  --header "Authorization: Bearer $MCP_BEARER_TOKEN"

# 3. 在会话里
# > 现在帮我巡检一次所有服务，挂的标红
# Claude 调用 full_report()，返回 Markdown 报告

# 4. 抓包验证协议正确性
npx @modelcontextprotocol/inspector --transport http \
  --url http://localhost:8000/mcp \
  --header "Authorization: Bearer $MCP_BEARER_TOKEN"

如果想用现成的脚手架而非手写，Claude Code 官方提供 mcp-server-dev 插件：

/plugin install mcp-server-dev@claude-plugins-official
/mcp-server-dev:build-mcp-server

它会问你用例、远程还是本地，自动生成 Streamable HTTP 或 stdio 骨架。

安全与运维清单

训练营适用性评估

维度 评估
适合训练营类型 云原生 / DevOps / 后端工程 / AI 平台工程训练营；MCP 进阶专题
学员基础要求 已完成 stdio MCP 入门 + 会一点 Python web
建议教学时长 3–4 小时（含部署到云主机 + 全员联调）
分组建议 2–3 人一组，每组部署一个不同主题的远程 MCP（巡检、值班、CI 状态），互相用 token 接入对方的服务，体验「服务方与消费方」角色
评估标准 服务可通过 URL + token 被任意组员接入；Inspector 抓包协议正确；nginx SSE 配置生效；能讲清 stdio 与 Streamable HTTP 的取舍

延伸阅读

参考来源

  1. Connect Claude Code to tools via MCP — Claude Code Docs — 官方 claude mcp add --transport http 命令、.mcp.json 的 streamable-http 类型与环境变量展开、远程鉴权与 OAuth 用法
  2. HTTP Deployment — FastMCP Documentation — run(transport="http") 与 http_app() ASGI 模式、自定义端点路径、无状态模式、nginx 反代与 SSE 缓冲配置
  3. MCP Streamable HTTP Transport: From SSE Migration to Production — apigene.ai — SSE 到 Streamable HTTP 的迁移路径、生产部署模式、单端点与按需流式的设计动机
  4. Understanding MCP Server Transports: STDIO, SSE, and HTTP Streamable — DEV Community — 三种传输的横向对比、何时该选 stdio / 何时该上云的决策框架
  5. MCP Specification 2025-06-18 — Tool Annotations — title / readOnlyHint / destructiveHint / idempotentHint / openWorldHint 五个注解字段的语义与客户端使用约定

[!tip] 51学通信知识星球 星主：爱卫生（微信：gprshome201101）——公众号「51学通信」创办者，19 年通信行业运维与讲师经验，著有《GPRS网络信令实例详解》《EPC原理与实践》《5G核心网原理与实践》三本纸质书（各大电商平台有售）。 星球定位：深耕 5G 核心网、IMS、NFV/AI、云原生等 ICT 领域，主打运维岗位与信令分析。 当前规模：5200 篇主题 / 990 个回答，198 元/年。 加入福利：加入星球后请加爱卫生微信，拉您进知识星球年度陪伴与答疑群——不限量回复您的各种问题，另有群内分享。 AI 加持：51学通信知识星球已作为 skill 接入 WorkBuddy，使用 deepseek-4-flash 在海量星球帖子中精准找贴、找视频、找 FAQ 和案例，做总结、找文档和 PPT 等场景全覆盖。

51学通信知识星球海报-扫码加入

📎 本分类最新文章