传输、安全、版本与发布

stdio 和 HTTP 的资源责任

stdio Client 拥有子进程生命周期，退出上下文应关闭管道并回收进程。远程 HTTP Server 独立存在，客户端断开不等于服务器任务取消。旧 HTTP+SSE 与 Streamable HTTP 不是简单改路径：会话、请求方法与流返回形状不同，必须两端协商同一版本。

固定 v1.12.4，在第一章 server.py 的运行行改为 mcp.run(transport='streamable-http')，在一个终端执行 python server.py。默认本地入口是 http://127.0.0.1:8000/mcp。另一个终端在相同环境保存 http_client.py：

import asyncio
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client
async def main():
    async with streamablehttp_client('http://127.0.0.1:8000/mcp') as (read,write,get_session_id):
        async with ClientSession(read,write) as session:
            await session.initialize()
            result=await session.call_tool('stock',{'sku':'A'})
            assert not result.isError and result.content[0].text=='3'
            print('HTTP stock',result.content[0].text)
asyncio.run(main())

这是本地无认证教学服务，只绑定开发环境，不应直接公开到公网。验证后在服务终端 Ctrl+C。某些新版 SDK 将函数改为 streamable_http_client 或改变返回值数量，必须按固定版本核对，不能把解包错误误判成网络故障。

协议兼容不能替代安全边界

远程认证需要验证 token 受众、资源与权限，不能把上游用户 token 无条件透传给任意下游。公开 OAuth 元数据和合法重定向也是身份流程的一部分。HTTP 服务验证 Origin 与主机绑定，避免浏览器跨站访问本机服务；工具调用仍做业务授权。

供应链控制包括固定启动命令、包版本与允许的环境变量。不要根据不可信 Server 的说明自动安装软件或泄露环境变量。工具说明与注解由 Server 提供，只能作为待信任数据。日志记录服务别名、方法、调用 ID、耗时与错误层级，不记录 bearer token。

一次升级需要对照哪些观察

保留上一版本虚拟环境；新环境安装候选版本，按顺序执行 initialize、list、合法调用、非法参数、业务失败、资源读取、提示获取、断线与关闭。协议版本、SDK 版本和 Host 产品支持是三条轴：SDK 支持某能力不代表所有 Host UI 都能展示它。

v1 FastMCP 与 v2 MCPServer 的迁移需要重新核对 import、客户端 transport 返回签名和依赖要求。旧 SSE 服务可保持独立维护直到客户端完成迁移，不能为了统一示例而把旧端点静默转发成不兼容的新协议。

发布时先用不调用模型的协议测试定位服务问题，再接模型测工具选择。将“协议可调用”与“模型会正确选择”分开，才能解释是 schema 描述、模型规划还是服务实现失败。

参阅 传输 与 授权。

当前 v2：在新环境重新建立连接

v1 实验用于理解现有集成；新项目可以采用当前 v2。不要在前面虚拟环境中覆盖依赖。另建目录并安装 mcp==2.2.0，Python 3.12 下执行：

python3.12 -m venv .venv
source .venv/bin/activate
python -m pip install 'mcp==2.2.0'

保存新的 server.py：

from mcp.server import MCPServer
from pydantic import BaseModel
class StockResult(BaseModel):
    found: bool
    count: int | None
mcp=MCPServer('inventory-v2')
@mcp.tool()
def stock(sku:str)->StockResult:
    values={'A':3,'B':0}
    return StockResult(found=sku in values,count=values.get(sku))
if __name__=='__main__':mcp.run(transport='stdio')

保存新的 client.py，运行 python client.py：

import asyncio,sys
from pathlib import Path
from mcp import Client,StdioServerParameters
async def main():
    params=StdioServerParameters(command=sys.executable,
        args=[str(Path(__file__).with_name('server.py'))])
    async with Client(params) as client:
        result=await client.call_tool('stock',{'sku':'B'})
        assert result.structured_content=={'found':True,'count':0}
        print(result.structured_content)
asyncio.run(main())

v2 的 Client 上下文负责连接与协商；字段采用 snake_case，例如 structured_content。不要把 v1 的 isError、inputSchema 直接当作 v2 Python 属性；线上的 JSON 字段与 Python 属性也不是同一层。v2 transport 的同步 handler 可能在线程池运行，依赖当前事件循环的同步函数需要改成 async handler。

当前协议演进还涉及能力发现、错误类型和旧 logging/roots/sampling 接口的弃用。迁移先保留 v1 可运行测试，再验证同一业务输入在 v2 的结果，不靠批量改 import 推断完成。细节见 v2 迁移指南。

让认证失败实际挡在工具前

下面使用MCP1.12.4的TokenVerifier与AuthSettings，保存auth_lab.py运行。真实HTTP请求经SDK认证中间件再到工具，传输使用本地ASGITransport，不建立TCP连接。演示Bearer只是固定测试值，绝不能作为线上令牌；没有模拟OAuth授权码流程。

import asyncio,time
import httpx
from mcp.server.fastmcp import FastMCP
from mcp.server.auth.provider import AccessToken
from mcp.server.auth.settings import AuthSettings
class DemoVerifier:
    async def verify_token(self,token):
        if token=='demo-reader':return AccessToken(token=token,client_id='reader',scopes=['read'],expires_at=int(time.time())+60)
        if token=='demo-noscope':return AccessToken(token=token,client_id='other',scopes=[],expires_at=int(time.time())+60)
        return None
mcp=FastMCP('protected',token_verifier=DemoVerifier(),stateless_http=True,json_response=True,
    auth=AuthSettings(issuer_url='https://issuer.invalid',resource_server_url='http://127.0.0.1:8000/mcp',required_scopes=['read']))
calls=[]
@mcp.tool()
def public_stock()->int:
    calls.append('read');return 3
async def main():
    app=mcp.streamable_http_app()
    async with app.router.lifespan_context(app):
        async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app),base_url='http://127.0.0.1:8000') as c:
            headers={'accept':'application/json, text/event-stream'}
            body={'jsonrpc':'2.0','id':1,'method':'tools/call','params':{'name':'public_stock','arguments':{}}}
            for token,expected in [(None,401),('bad',401),('demo-noscope',403),('demo-reader',200)]:
                h=dict(headers)
                if token:h['authorization']='Bearer '+token
                r=await c.post('/mcp',headers=h,json=body)
                assert r.status_code==expected,(r.status_code,r.text)
            assert calls==['read']
    print('401,403 and authorized tool boundary passed')
asyncio.run(main())

未带令牌或无效令牌返回401；身份有效但缺read scope返回403；只有reader让calls增加一次。生产TokenVerifier必须验证授权服务器签名、issuer、audience/resource、期限和撤销策略，不能只检查令牌非空。OAuth发现、PKCE和用户授权由真实授权服务器及客户端实现，resource server负责验证其发行的访问令牌。本例issuer.invalid仅使配置形状明确，不能用于真实授权。

ASGI测试直接发送工具请求以定位认证中间件；完整MCP客户端的initialize与能力协商仍按第一章执行。工具内部的数据资源权限仍需用受信任身份复验，read scope不等于所有租户均可读。

练习

把 http_client 的 sku 改为 B，应返回 "0"；然后停止服务再次运行客户端，必须出现连接异常，不能打印成功。说明两个观察分别证明什么。

先记录自己的输出与解释，再核对本章答案。