MCP Streamable HTTP 远程部署实战：从 OAuth 2.1 认证到公网访问

本地 stdio 跑通的 MCP Server，搬到公网就变成另一个故事。2026 年 MCP 协议把 Streamable HTTP 定为唯一的推荐远程传输，并强制要求远程服务器接入 OAuth 2.1。本文用 Python 完整实现一个带认证、可公网访问的远程 MCP Server，并跑通客户端连接。

为什么本地 MCP 跑得好，上公网就麻烦

绝大多数 MCP 教程停在 stdio 传输：客户端通过子进程启动 server，用标准输入输出通信。这套机制有三个硬限制：

• 只能本地：stdio 是进程间通信，跨机器没法用
• 单客户端独占：一个 server 进程只能服务一个客户端
• 没有鉴权：谁起进程谁能用，公网暴露等于裸奔

要把 MCP Server 真正提供给团队或外部使用，必须换成 HTTP 传输。2025 年底 MCP 协议淘汰了旧的 HTTP+SSE 双端点方案，统一为 Streamable HTTP：单一端点，既支持普通请求响应，也支持 SSE 流式推送，状态由会话 ID 维护。同时协议规定，远程 MCP Server 必须实现 OAuth 2.1 授权流程。

Streamable HTTP 传输协议要点

Streamable HTTP 的核心设计：

• 单端点：所有通信走 /mcp 一个 URL，通过 Accept 头区分响应类型
• 会话可选：无状态模式每次请求独立；有状态模式用 Mcp-Session-Id 头维系会话
• 流式响应：客户端发 Accept: text/event-stream，服务器可返回 SSE 流，适合长任务进度推送
• 初始化握手：客户端发 initialize 请求，服务器返回 session ID，后续请求带上这个 ID

远程部署的安全要求：

• 必须实现 OAuth 2.1，支持 Authorization Code + PKCE 流程
• Token 端点返回 RFC 8707 的 resource 参数，绑定具体 server
• 客户端在 MCP 请求头里带 Authorization: Bearer <token>

实战：搭建带 OAuth 2.1 的远程 MCP Server

下面用 Python 的 mcp SDK 配合 FastAPI 搭一个"企业内部知识库查询"的远程 Server，提供两个工具：搜索文档、获取文档详情。

项目结构

remote-mcp-server/
├── server.py          # MCP Server 主程序
├── auth.py            # OAuth 2.1 授权服务
├── requirements.txt
└── .env

安装依赖

pip install "mcp[cli]" fastapi uvicorn itsdangerous httpx

MCP Server 主程序

server.py 用 Streamable HTTP 传输，提供两个文档检索工具：

import os
from mcp.server import Server
from mcp.server.fastmcp import FastMCP
from mcp.types import Tool, TextContent

# 模拟企业知识库
DOCS = {
    "doc-001": {"title": "退款流程规范", "content": "用户发起退款后 24 小时内审核..."},
    "doc-002": {"title": "API 限流策略", "content": "单 IP 每分钟 60 次请求..."},
    "doc-003": {"title": "数据备份方案", "content": "每日全量 + 每小时增量备份..."},
}

mcp = FastMCP("enterprise-kb")

@mcp.tool()
def search_docs(keyword: str) -> str:
    """按关键词搜索企业知识库文档

    Args:
        keyword: 搜索关键词
    """
    results = []
    for doc_id, doc in DOCS.items():
        if keyword in doc["title"] or keyword in doc["content"]:
            results.append(f"[{doc_id}] {doc['title']}")
    return "\n".join(results) if results else "未找到匹配文档"

@mcp.tool()
def get_doc_detail(doc_id: str) -> str:
    """获取指定文档的完整内容

    Args:
        doc_id: 文档 ID，如 doc-001
    """
    doc = DOCS.get(doc_id)
    if not doc:
        return f"文档 {doc_id} 不存在"
    return f"标题：{doc['title']}\n内容，{doc['content']}"

if __name__ == "__main__":
    mcp.run(transport="streamable-http", host="0.0.0.0", port=8000)

OAuth 2.1 授权服务

远程 MCP Server 要求客户端先拿 token。auth.py 实现一个最小可用的 OAuth 2.1 授权服务器：

import os
import secrets
import time
from fastapi import FastAPI, Request, Form
from fastapi.responses import RedirectResponse, JSONResponse

auth_app = FastAPI(title="MCP OAuth 2.1 Server")

# 内存存储（生产环境请换数据库）
CLIENTS = {"mcp-cli-client": {"redirect_uris": ["http://localhost:3000/callback"]}}
AUTH_CODES = {}     # code -> {client_id, user, resource}
TOKENS = {}         # token -> {client_id, resource, exp}
PKCE_CHALLENGES = {}  # code_verifier -> challenge

RESOURCE = "https://kb.example.com"  # 本 MCP Server 的资源标识

@auth_app.get("/.well-known/oauth-authorization-server")
def metadata():
    return {
        "issuer": "https://auth.example.com",
        "authorization_endpoint": "https://auth.example.com/authorize",
        "token_endpoint": "https://auth.example.com/token",
        "registration_endpoint": "https://auth.example.com/register",
        "scopes_supported": ["mcp"],
        "response_types_supported": ["code"],
        "grant_types_supported": ["authorization_code", "refresh_token"],
        "code_challenge_methods_supported": ["S256"],
        "token_endpoint_auth_methods_supported": ["none"],
    }

@auth_app.get("/authorize")
def authorize(
    response_type: str,
    client_id: str,
    redirect_uri: str,
    code_challenge: str,
    code_challenge_method: str,
    resource: str,
    state: str,
):
    # 简化版：直接签发授权码，生产环境这里要做用户登录
    if client_id not in CLIENTS:
        return JSONResponse({"error": "invalid_client"}, status_code=400)
    code = secrets.token_urlsafe(32)
    AUTH_CODES[code] = {
        "client_id": client_id,
        "resource": resource,
        "exp": time.time() + 600,
    }
    return RedirectResponse(
        f"{redirect_uri}?code={code}&state={state}"
    )

@auth_app.post("/token")
def token(
    grant_type: str = Form(...),
    code: str = Form(None),
    redirect_uri: str = Form(None),
    client_id: str = Form(...),
    code_verifier: str = Form(None),
):
    if grant_type != "authorization_code":
        return JSONResponse({"error": "unsupported_grant_type"}, status_code=400)

    record = AUTH_CODES.pop(code, None)
    if not record or record["exp"] < time.time():
        return JSONResponse({"error": "invalid_grant"}, status_code=400)

    access_token = secrets.token_urlsafe(32)
    TOKENS[access_token] = {
        "client_id": client_id,
        "resource": record["resource"],
        "exp": time.time() + 3600,
    }
    return {
        "access_token": access_token,
        "token_type": "Bearer",
        "expires_in": 3600,
        "resource": record["resource"],
    }

把鉴权接到 MCP Server

MCP SDK 支持 Starlette 中间件，在请求进入 server 前校验 Bearer Token：

from starlette.applications import Starlette
from starlette.middleware import Middleware
from starlette.requests import Request
from starlette.responses import JSONResponse
from mcp.server.streamable_http_manager import StreamableHTTPSessionManager

async def auth_middleware(request: Request, call_next):
    # 排除 OAuth 元数据和健康检查
    if request.url.path.startswith("/.well-known"):
        return await call_next(request)

    auth_header = request.headers.get("Authorization", "")
    if not auth_header.startswith("Bearer "):
        return JSONResponse({"error": "missing_token"}, status_code=401)
    token = auth_header[7:]
    record = TOKENS.get(token)
    if not record or record["exp"] < time.time():
        return JSONResponse({"error": "invalid_token"}, status_code=401)
    request.state.client_id = record["client_id"]
    return await call_next(request)

部署到公网

本地跑通后，公网部署推荐两种方式。

方式一：Nginx 反向代理 + HTTPS

MCP 协议要求远程传输必须走 HTTPS。用 Nginx 做反代，把 kb.example.com 指向本地的 8000 端口：

server {
    listen 443 ssl http2;
    server_name kb.example.com;

    ssl_certificate     /etc/letsencrypt/live/kb.example.com/fullchain.pem;
    ssl_certificate_key /etc/letsencrypt/live/kb.example.com/privkey.pem;

    location / {
        proxy_pass http://127.0.0.1:8000;
        proxy_http_version 1.1;
        proxy_set_header Host $host;
        proxy_set_header Connection "";

        # SSE 长连接支持
        proxy_buffering off;
        proxy_cache off;
        proxy_read_timeout 86400s;
    }
}

proxy_buffering off 是关键：SSE 流式响应必须关闭缓冲，否则 Nginx 会攒满 buffer 才转发，长任务进度就卡住了。

用 systemd 管理 server 进程：

# /etc/systemd/system/mcp-kb.service
[Unit]
Description=MCP Knowledge Base Server
After=network.target

[Service]
Type=simple
User=www-data
WorkingDirectory=/opt/remote-mcp-server
ExecStart=/opt/remote-mcp-server/.venv/bin/python server.py
Restart=always
RestartSec=5

[Install]
WantedBy=multi-user.target

方式二：Cloudflare Tunnel 免公网 IP

没有公网 IP 的机器，用 Cloudflare Tunnel 把本地服务暴露到公网：

# 安装 cloudflared 后
cloudflared tunnel create mcp-kb
cloudflared tunnel route dns mcp-kb kb.example.com

# 配置 ~/.cloudflared/config.yml
cat > ~/.cloudflared/config.yml <<EOF
tunnel: mcp-kb
credentials-file: ~/.cloudflared/<tunnel-id>.json
ingress:
  - hostname: kb.example.com
    service: http://localhost:8000
  - service: http_status:404
EOF

cloudflared tunnel run mcp-kb

Cloudflare 自动处理 HTTPS 证书和 SSE 长连接，适合快速验证。

客户端连接测试

Claude Code 连接远程 MCP

在项目根目录的 .mcp.json 里配置远程 server：

{
  "mcpServers": {
    "enterprise-kb": {
      "url": "https://kb.example.com/mcp",
      "type": "http"
    }
  }
}

首次连接时 Claude Code 会自动走 OAuth 2.1 流程：弹出浏览器登录授权，拿到 token 后缓存到本地 keychain，后续请求自动带 Bearer Token。

Python 客户端测试脚本

client.py 用 MCP SDK 直连，跳过 OAuth 用预设 token 测试：

import asyncio
from mcp.client.session import ClientSession
from mcp.client.streamable_http import streamablehttp_client

async def main():
    headers = {"Authorization": "Bearer <你的 access_token>"}
    async with streamablehttp_client(
        "https://kb.example.com/mcp", headers=headers
    ) as (read, write, _):
        async with ClientSession(read, write) as session:
            await session.initialize()
            # 列出可用工具
            tools = await session.list_tools()
            for t in tools.tools:
                print(f"工具，{t.name}，{t.description}")
            # 调用搜索
            result = await session.call_tool(
                "search_docs", {"keyword": "退款"}
            )
            print("搜索结果：", result.content[0].text)

asyncio.run(main())

正常输出：

工具，search_docs，按关键词搜索企业知识库文档
工具，get_doc_detail，获取指定文档的完整内容
搜索结果： [doc-001] 退款流程规范

常见问题 FAQ

Q1：Streamable HTTP 和旧的 HTTP+SSE 有什么区别？
旧的方案要两个端点（POST 提交请求 + GET 订阅 SSE），状态管理复杂；Streamable HTTP 合并成单个 /mcp 端点，靠 Accept 头决定返回普通 JSON 还是 SSE 流，实现更简单，也更容易过 CDN 和反代。

Q2：OAuth 2.1 一定要自己搭授权服务器吗？
不一定。可以用 Auth0、Authentik、Keycloak 等现成的 IdP，只要它们支持 RFC 8707 的 resource 参数即可。本文自己实现是为了讲清原理，生产环境直接接现成的 IdP 更省事。

Q3：SSE 流式响应在 Nginx 后面经常断，怎么排查？
先确认 proxy_buffering off 和 proxy_cache off 都配了；再检查 proxy_read_timeout，默认 60 秒对长任务太短，调到 86400s；最后看 Cloudflare 之类的 CDN 有没有自己的超时设置，免费版 CDN 对长连接支持有限。

Q4：一个远程 MCP Server 能同时服务多个客户端吗？
能。Streamable HTTP 是无状态优先的，每个请求独立处理；需要会话时用 Mcp-Session-Id 区分，多个 session 并行不冲突。这比 stdio 的单客户端独占模式强很多。

Q5：Claude Code 连远程 server 一直报 401，怎么调试？
先用 curl -H "Authorization: Bearer <token>" https://kb.example.com/mcp 手动验证 token 是否有效；再检查 token 的 resource 字段是否和 server 配置的 RESOURCE 一致；最后看 server 日志里中间件有没有拒绝请求。MCP SDK 的报错信息比较简略，手动 curl 能快速定位是鉴权还是传输层的问题。

小结

把 MCP Server 从本地 stdio 搬到公网，本质上是从"进程间通信"切换到"带鉴权的 HTTP 服务"。Streamable HTTP 用单端点 + 会话 ID 把传输层简化了，OAuth 2.1 则补上了安全这道门。本地原型验证用 Cloudflare Tunnel 最快，正式生产用 Nginx 反代 + systemd 托管更稳。协议层面理解了"单端点、会话可选、流式响应"这三点，远程 MCP 的部署就不再是黑盒。

本文发布于 1630.top，转载请注明出处。

本文发布于 1630.top，转载请注明出处。