第九章：MCP与工具生态集成¶

CLIENT_META = {
    "io.modelcontextprotocol/protocolVersion": "2026-07-28",
    "io.modelcontextprotocol/clientCapabilities": {
        "elicitation": {}
    },
    "io.modelcontextprotocol/clientInfo": {
        "name": "HarnessMCPClient",
        "version": "1.0.0"
    }
}

def build_request(method: str, params: dict, request_id: int) -> dict:
    """所有 MCP 请求统一在这里构造，确保 _meta 不会漏发"""
    return {
        "jsonrpc": "2.0",
        "id": request_id,
        "method": method,
        "params": {**params, "_meta": dict(CLIENT_META)}
    }

async def discover(server: MCPServer) -> dict:
    request = build_request("server/discover", {}, generate_request_id())
    result = (await server.call(request))["result"]

    return {
        "versions": result["supportedVersions"],     # 支持的协议版本列表
        "capabilities": result["capabilities"],      # Server能力
        "instructions": result.get("instructions"),  # 可选的使用说明
        "ttl_ms": result["ttlMs"],                   # 本结果的新鲜度提示（毫秒）
        "cache_scope": result["cacheScope"],         # "public" 或 "private"
        "info": result["_meta"]["io.modelcontextprotocol/serverInfo"],
    }

# 仅用于回退到 2025-11-25 旧版Server的兼容路径
init_message = {
    "jsonrpc": "2.0",
    "id": 0,
    "method": "initialize",
    "params": {
        "protocolVersion": "2025-11-25",
        "capabilities": {
            "sampling": {},
            "roots": {"listChanged": True},
            "elicitation": {}
        },
        "clientInfo": {
            "name": "HarnessMCPClient",
            "version": "1.0.0"
        }
    }
}

class ToolSchemaCache:
    def __init__(self, cache_path="~/.harness/tool_cache.db"):
        # sqlite3 不会展开 ~，需要 expanduser
        self.db = sqlite3.connect(os.path.expanduser(cache_path))
        self.db.execute("""
            CREATE TABLE IF NOT EXISTS tool_schemas (
                server_name TEXT,
                tool_name TEXT,
                schema TEXT,
                schema_version TEXT,
                cached_at INTEGER,
                PRIMARY KEY (server_name, tool_name)
            )
        """)

    def get_or_fetch(self, server: MCPServer, tool_name: str) -> dict:
        # 1. 先查缓存
        cached = self.db.execute(
            "SELECT schema, schema_version FROM tool_schemas WHERE server_name = ? AND tool_name = ?",
            (server.name, tool_name)
        ).fetchone()

        if cached:
            schema_json, cached_version = cached

            # 2. 检查本地缓存版本是否过期
            current_version = server.get_cached_schema_version(tool_name)
            if current_version == cached_version:
                return json.loads(schema_json)  # 命中缓存

        # 3. 缓存未命中,重新获取
        schema = server.fetch_tool_schema(tool_name)
        self.db.execute(
            "INSERT OR REPLACE INTO tool_schemas VALUES (?, ?, ?, ?, ?)",
            (server.name, tool_name, json.dumps(schema),
             server.get_cached_schema_version(tool_name), int(time.time()))
        )
        self.db.commit()
        return schema

# 连接池配置
HTTP_CLIENT = httpx.AsyncClient(
    limits=httpx.Limits(max_connections=10, max_keepalive_connections=5),
    timeout=30.0
)

async def read_resource_from_server(server: MCPServer, uri: str) -> dict:
    """读取资源内容，兼容 application/json 与 text/event-stream 响应"""
    request = {
        "jsonrpc": "2.0",
        "id": generate_request_id(),
        "method": "resources/read",
        "params": {"uri": uri}
    }

    headers = {
        "Accept": "application/json, text/event-stream",
        "MCP-Protocol-Version": "2025-11-25",
    }
    if server.session_id:
        headers["MCP-Session-Id"] = server.session_id

    async with HTTP_CLIENT.stream("POST", server.endpoint, json=request, headers=headers) as response:
        response.raise_for_status()
        content_type = response.headers.get("content-type", "")
        if content_type.startswith("application/json"):
            return json.loads(await response.aread())

        if content_type.startswith("text/event-stream"):
            async for event in parse_sse_events(response.aiter_lines()):
                if event.get("data"):
                    message = json.loads(event["data"])
                    if message.get("id") == request["id"]:
                        return message

        raise ValueError(f"Unsupported MCP response type: {content_type}")

class ToolCallState(State):
    """工具调用状态"""

    def __init__(self, server_name: str, tool_name: str, params: dict):
        self.server = harness.get_mcp_server(server_name)
        self.tool_name = tool_name
        self.params = params

    async def execute(self, context: ExecutionContext) -> StateTransition:
        try:
            # 1. 准备工具调用请求
            request = {
                "jsonrpc": "2.0",
                "id": context.request_id,
                "method": "tools/call",
                "params": {
                    "name": self.tool_name,
                    "arguments": self.params
                }
            }

            # 2. 发送请求(带超时和重试)
            result = await self.server.call_with_retry(
                request,
                max_retries=3,
                timeout=30
            )

            # 3. 检查结果
            if "error" in result:
                return StateTransition(
                    next_state="error_handling",
                    context={"error": result["error"]}
                )

            # 4. 结果写入上下文
            context.set("tool_result", result["result"])

            return StateTransition(next_state="next_step")

        except TimeoutError:
            return StateTransition(
                next_state="retry_or_fallback",
                context={"reason": "timeout"}
            )

class AtomicToolExecution:
    """原子性工具执行"""

    def __init__(self, execution_id: str):
        self.execution_id = execution_id
        self.state_file = f"/tmp/harness/{execution_id}.state"

    async def execute(self, server: MCPServer, request: dict) -> dict:
        # 1. 检查是否已经执行过(幂等性)
        if os.path.exists(self.state_file):
            with open(self.state_file) as f:
                cached_result = json.load(f)
            return cached_result  # 重复执行,直接返回缓存

        # 2. 执行工具调用
        result = await server.call(request)

        # 3. 原子地写入状态(保证不会丢失)
        os.makedirs(os.path.dirname(self.state_file), exist_ok=True)
        with open(self.state_file + ".tmp", "w") as f:
            json.dump(result, f)
        os.rename(self.state_file + ".tmp", self.state_file)  # 原子rename

        return result

async def call_tool(server_name: str, tool_name: str, params: dict) -> dict:
    # 1. 检查权限策略
    if not harness.policy_engine.is_allowed(server_name, tool_name, params):
        raise PermissionDenied(f"Tool call denied by policy: {server_name}/{tool_name}")

    # 2. 执行工具调用
    return await harness.mcp_servers[server_name].call(tool_name, params)

import json
import subprocess
from typing import Any, Callable, Dict, Optional

PROTOCOL_VERSION = "2026-07-28"
META_VERSION = "io.modelcontextprotocol/protocolVersion"
META_CAPABILITIES = "io.modelcontextprotocol/clientCapabilities"
META_CLIENT_INFO = "io.modelcontextprotocol/clientInfo"
META_LOG_LEVEL = "io.modelcontextprotocol/logLevel"
META_SERVER_INFO = "io.modelcontextprotocol/serverInfo"


class StdioMCPClient:
    """基于stdio的MCP客户端（2026-07-28 无状态模型）

    没有 initialize 握手，每个请求自带完整信封。子进程可以活很久，
    但它不是会话：Server 不得从同一管道上的历史请求推断任何状态。
    """

    def __init__(self, server_path: str, server_args: list = None):
        """启动MCP Server进程"""
        self.server_path = server_path
        self.server_args = server_args or []
        self.process: Optional[subprocess.Popen] = None
        self.request_id = 0
        # 仅客户端侧的本地缓存，属于优化，不构成协议状态
        self.server_info: Optional[Dict[str, Any]] = None

    def start(self) -> None:
        """启动Server进程。stderr 不重定向，Server 日志直接透传到父进程，
        免得管道写满后把子进程堵死"""
        self.process = subprocess.Popen(
            [self.server_path] + self.server_args,
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            text=True,
            bufsize=1,  # 行缓冲
        )

    def _envelope(self, log_level: Optional[str] = None) -> Dict[str, Any]:
        """构造无状态请求信封。前两个键是必需的，逐请求重复携带是设计而非冗余；
        stdio 上没有 HTTP 头可用，信封只能落在 params._meta 里"""
        meta: Dict[str, Any] = {
            META_VERSION: PROTOCOL_VERSION,
            META_CAPABILITIES: {},
            META_CLIENT_INFO: {"name": "stdio-example", "version": "0.1.0"},
        }
        if log_level:
            # 日志级别也逐请求携带，取代已移除的 logging/setLevel
            meta[META_LOG_LEVEL] = log_level
        return meta

    def send_request(
        self,
        method: str,
        params: Dict[str, Any] = None,
        log_level: str = None,
        on_notification: Callable[[Dict[str, Any]], None] = None,
    ) -> Dict[str, Any]:
        """发送JSON-RPC请求并等待响应"""
        if not self.process:
            raise RuntimeError("Server not started")

        self.request_id += 1
        request_id = self.request_id

        # 每个请求都要注入信封；调用方自带的 _meta（如 progressToken）优先保留
        payload = dict(params or {})
        payload["_meta"] = {**self._envelope(log_level), **payload.get("_meta", {})}
        request = {
            "jsonrpc": "2.0",
            "id": request_id,
            "method": method,
            "params": payload,
        }

        # 写入请求
        self.process.stdin.write(json.dumps(request) + "\n")
        self.process.stdin.flush()

        # 读取响应：stdout 上会混进属于本请求的通知（进度、日志），
        # 必须读到同 id 的那条响应才算收工
        while True:
            response_line = self.process.stdout.readline()
            if not response_line:
                raise RuntimeError("Server closed connection")
            message = json.loads(response_line)
            if message.get("id") == request_id:
                # Server 在每个结果的 _meta 里回带 serverInfo，取代握手时的一次性自述
                meta = (message.get("result") or {}).get("_meta") or {}
                if META_SERVER_INFO in meta:
                    self.server_info = meta[META_SERVER_INFO]
                return message
            if "method" in message and "id" not in message:
                if on_notification:
                    on_notification(message)
                continue
            # 既不是本请求的响应，也不是通知。2026-07-28 的 Server 不会在这里反向
            # 发起请求：要客户端补输入时改走 MRTR，即在响应里返回 input_required
            raise RuntimeError(f"Unexpected message: {message}")

    def send_notification(self, method: str, params: Dict[str, Any] = None) -> None:
        """发送JSON-RPC通知，不等待响应"""
        if not self.process:
            raise RuntimeError("Server not started")
        notification = {"jsonrpc": "2.0", "method": method}
        if params:
            notification["params"] = params
        self.process.stdin.write(json.dumps(notification) + "\n")
        self.process.stdin.flush()

    def cancel(self, request_id: int, reason: str = "") -> None:
        """取消在途请求。HTTP 上关掉 SSE 流就是取消信号，
        stdio 上没有流可关，只能显式发 notifications/cancelled"""
        self.send_notification(
            "notifications/cancelled",
            {"requestId": request_id, "reason": reason},
        )

    def _unwrap(self, response: Dict[str, Any]) -> Dict[str, Any]:
        """取出 result，JSON-RPC 错误直接抛出"""
        if "error" in response:
            error = response["error"]
            raise RuntimeError(f"JSON-RPC error {error['code']}: {error['message']}")
        return response["result"]

    def discover(self) -> Optional[Dict[str, Any]]:
        """探测Server，一次拿到 supportedVersions 与 capabilities

        stdio 上它同时是判断对端新旧的探针：只讲握手协议的 Server 不认识这个方法，
        会回 -32601，此时返回 None，由调用方决定是否退回 2025-11-25 兼容路径。
        """
        response = self.send_request("server/discover")
        error = response.get("error")
        if error:
            if error.get("code") == -32601:
                return None
            raise RuntimeError(f"discover failed: {error}")
        return response["result"]

    def list_tools(self) -> Dict[str, Any]:
        """列出工具。结果自带 cacheScope/ttlMs，缓存放在客户端，Server 依然无状态"""
        return self._unwrap(self.send_request("tools/list"))

    def call_tool(
        self,
        tool_name: str,
        arguments: Dict[str, Any],
        progress_token: str = None,
        log_level: str = None,
        on_notification: Callable[[Dict[str, Any]], None] = None,
    ) -> Dict[str, Any]:
        """调用工具"""
        params: Dict[str, Any] = {
            "name": tool_name,
            "arguments": arguments,
        }
        if progress_token:
            # 进度这类请求内通知只走本请求的往返，不需要常驻订阅
            params["_meta"] = {"progressToken": progress_token}
        return self._unwrap(
            self.send_request(
                "tools/call",
                params,
                log_level=log_level,
                on_notification=on_notification,
            )
        )

    def close(self) -> None:
        """关闭连接"""
        if self.process:
            self.process.terminate()
            self.process.wait()

    def __enter__(self):
        self.start()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.close()

# stdio传输使用示例
if __name__ == "__main__":
    with StdioMCPClient("/path/to/mcp_server") as client:
        # 先探测：拿到能力清单，同时确认对端讲的是无状态版本
        info = client.discover()
        if info is None:
            raise SystemExit("对端仍在握手时代，需退回 2025-11-25 兼容路径")
        print("supportedVersions:", info["supportedVersions"])
        print("serverInfo:", client.server_info)

        # 列出可用工具：每个请求各自完备，不依赖上一次请求留下的痕迹
        listed = client.list_tools()
        print("tools:", [tool["name"] for tool in listed["tools"]])
        print("cache:", listed.get("cacheScope"), listed.get("ttlMs"))

        # 调用工具，顺带接收只属于这个请求的进度通知
        result = client.call_tool(
            "add",
            {"a": 2, "b": 3},
            progress_token="call-1",
            log_level="info",
            on_notification=lambda n: print("  notify:", n["method"], n["params"]),
        )
        print("resultType:", result["resultType"], "content:", result["content"])

import aiohttp
import json
import asyncio
from typing import Dict, Any, List, Optional, AsyncIterator

PROTOCOL_VERSION = "2026-07-28"

# 无状态请求信封的保留键，全部放在 params._meta 里
PROTOCOL_VERSION_KEY = "io.modelcontextprotocol/protocolVersion"
CLIENT_CAPABILITIES_KEY = "io.modelcontextprotocol/clientCapabilities"
CLIENT_INFO_KEY = "io.modelcontextprotocol/clientInfo"
SERVER_INFO_KEY = "io.modelcontextprotocol/serverInfo"
SUBSCRIPTION_ID_KEY = "io.modelcontextprotocol/subscriptionId"

# 需要额外携带 Mcp-Name 头的方法 → 该值所在的参数名
NAME_BEARING_METHODS = {
    "tools/call": "name",
    "prompts/get": "name",
    "resources/read": "uri",
}

PARSE_ERROR = -32700
INVALID_REQUEST = -32600
METHOD_NOT_FOUND = -32601
INVALID_PARAMS = -32602
HEADER_MISMATCH = -32020
UNSUPPORTED_PROTOCOL_VERSION = -32022

# JSON-RPC错误码到HTTP状态的映射表，两端读同一张表
ERROR_HTTP_STATUS = {
    PARSE_ERROR: 400,
    INVALID_REQUEST: 400,
    INVALID_PARAMS: 400,
    HEADER_MISMATCH: 400,
    UNSUPPORTED_PROTOCOL_VERSION: 400,
    METHOD_NOT_FOUND: 404,
}


class MCPRequestError(RuntimeError):
    """服务端返回的JSON-RPC错误"""

    def __init__(self, code: int, message: str, data: Any = None):
        super().__init__(f"[{code}] {message}")
        self.code = code
        self.message = message
        self.data = data


class StreamableHttpMCPClient:
    """基于Streamable HTTP的MCP客户端"""

    def __init__(self, server_url: str):
        self.server_url = server_url.rstrip('/')
        self.request_id = 0
        self.session: Optional[aiohttp.ClientSession] = None
        self.client_info = {"name": "streamable-http-example", "version": "0.1.0"}
        self.client_capabilities: Dict[str, Any] = {}

    def _envelope(self, params: Optional[Dict[str, Any]]) -> Dict[str, Any]:
        """把请求信封写进 params._meta。没有握手，所以每个请求都要自带版本与能力。"""
        params = dict(params or {})
        meta = dict(params.get("_meta") or {})
        meta[PROTOCOL_VERSION_KEY] = PROTOCOL_VERSION
        meta[CLIENT_CAPABILITIES_KEY] = self.client_capabilities
        # clientInfo 是可选键，规范建议带上，服务端只当作展示信息
        meta[CLIENT_INFO_KEY] = self.client_info
        params["_meta"] = meta
        return params

    def _headers(self, method: str, params: Dict[str, Any]) -> Dict[str, str]:
        """Accept必须同时列出两种类型；三个MCP头让中间层无需解析body就能路由。"""
        headers = {
            "Accept": "application/json, text/event-stream",
            "Content-Type": "application/json",
            # 这个头的值必须与 _meta 里的协议版本完全一致
            "MCP-Protocol-Version": PROTOCOL_VERSION,
            "Mcp-Method": method,
        }
        name_key = NAME_BEARING_METHODS.get(method)
        if name_key is not None and isinstance(params.get(name_key), str):
            headers["Mcp-Name"] = params[name_key]
        return headers

    async def connect(self) -> None:
        """只准备HTTP连接池；没有initialize，第一个请求就是正常业务请求。"""
        self.session = aiohttp.ClientSession()

    def _build_request(self, method: str, params: Optional[Dict[str, Any]]) -> Dict[str, Any]:
        """构造一条带信封的JSON-RPC请求"""
        self.request_id += 1
        return {
            "jsonrpc": "2.0",
            "id": self.request_id,
            "method": method,
            "params": self._envelope(params),
        }

    @staticmethod
    def _unwrap(message: Dict[str, Any]) -> Dict[str, Any]:
        """取出result；错误响应连同HTTP 4xx一起抛成MCPRequestError。"""
        if "error" in message:
            error = message["error"]
            raise MCPRequestError(error.get("code"), error.get("message", ""), error.get("data"))
        return message.get("result", {})

    @staticmethod
    async def _iter_sse(resp: aiohttp.ClientResponse) -> AsyncIterator[Dict[str, Any]]:
        """把SSE流切成JSON-RPC消息；以冒号开头的注释行是保活心跳，直接丢弃。"""
        data_lines: List[str] = []
        async for raw_line in resp.content:
            line = raw_line.decode("utf-8").rstrip("\r\n")
            if not line:
                if data_lines:
                    yield json.loads("\n".join(data_lines))
                    data_lines = []
                continue
            if line.startswith(":"):
                continue
            field, _, value = line.partition(":")
            if field == "data":
                data_lines.append(value.removeprefix(" "))

    async def send_request(
        self, method: str, params: Dict[str, Any] = None, timeout: float = 30
    ) -> Dict[str, Any]:
        """一条JSON-RPC请求就是一次独立POST，请求之间不共享任何服务端状态。"""
        request = self._build_request(method, params)
        async with self.session.post(
            f"{self.server_url}/mcp",
            json=request,
            headers=self._headers(method, request["params"]),
            timeout=aiohttp.ClientTimeout(total=timeout),
        ) as resp:
            if resp.content_type == "text/event-stream":
                # 请求内SSE：流上先送本请求的进度、日志通知，最后一帧才是响应。
                # 这条流只服务这一个请求，提前关闭它就是取消信号。
                async for message in self._iter_sse(resp):
                    if message.get("id") == request["id"]:
                        return self._unwrap(message)
                    self.on_notification(message)
                raise MCPRequestError(INVALID_REQUEST, f"{method} 的响应流提前结束")
            if resp.content_type != "application/json":
                # 响应体无法识别成JSON-RPC，才是回落到旧版initialize握手的信号
                raise MCPRequestError(
                    INVALID_REQUEST, f"HTTP {resp.status}: {(await resp.text())[:120]}"
                )
            # 4xx的body同样是JSON-RPC错误对象：缺信封400/-32602、
            # 版本不支持400/-32022、未知方法404/-32601
            return self._unwrap(await resp.json())

    def on_notification(self, message: Dict[str, Any]) -> None:
        """请求内通知的处理钩子，子类覆盖它即可"""
        print(f"Notification: {message.get('method')}")

    async def listen(
        self,
        tools_list_changed: bool = False,
        prompts_list_changed: bool = False,
        resources_list_changed: bool = False,
        resource_subscriptions: Optional[List[str]] = None,
    ) -> AsyncIterator[Dict[str, Any]]:
        """订阅长期变更通知：POST subscriptions/listen，它的响应本身就是一条长连SSE。

        流上第一帧是确认，之后是订阅到的变更通知，服务端优雅关闭时以本请求的响应收尾。
        没有退订方法，跳出循环让流关闭就是退订。
        """
        notifications: Dict[str, Any] = {}
        if tools_list_changed:
            notifications["toolsListChanged"] = True
        if prompts_list_changed:
            notifications["promptsListChanged"] = True
        if resources_list_changed:
            notifications["resourcesListChanged"] = True
        if resource_subscriptions:
            notifications["resourceSubscriptions"] = list(resource_subscriptions)

        request = self._build_request("subscriptions/listen", {"notifications": notifications})
        async with self.session.post(
            f"{self.server_url}/mcp",
            json=request,
            headers=self._headers("subscriptions/listen", request["params"]),
            # 长连流不能设总超时，否则订阅会被客户端自己掐断
            timeout=aiohttp.ClientTimeout(total=None, sock_read=None),
        ) as resp:
            if resp.content_type != "text/event-stream":
                self._unwrap(await resp.json())
                raise MCPRequestError(INVALID_REQUEST, "subscriptions/listen 的响应不是SSE流")
            async for message in self._iter_sse(resp):
                if message.get("id") == request["id"]:
                    return  # 服务端优雅关闭
                yield message

    async def call_tool(self, tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
        """调用工具"""
        return await self.send_request(
            "tools/call",
            {"name": tool_name, "arguments": arguments}
        )

    async def close(self) -> None:
        """关闭连接"""
        if self.session:
            await self.session.close()

    async def __aenter__(self):
        await self.connect()
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.close()

# Server端实现示例
from aiohttp import web
import asyncio

class StreamableHttpMCPServer:
    """基于Streamable HTTP的MCP服务器"""

    SUPPORTED_VERSIONS = [PROTOCOL_VERSION]

    # 变更通知方法 → 订阅过滤器里的开关名
    FILTER_KEYS = {
        "notifications/tools/list_changed": "toolsListChanged",
        "notifications/prompts/list_changed": "promptsListChanged",
        "notifications/resources/list_changed": "resourcesListChanged",
    }

    def __init__(
        self,
        host: str = "127.0.0.1",
        port: int = 8000,
        bearer_token: Optional[str] = None,
        allowed_origins: Optional[set] = None,
    ):
        self.host = host
        self.port = port
        self.bearer_token = bearer_token
        self.allowed_origins = allowed_origins or {"http://localhost:8000"}
        self.subscribers: List[Dict[str, Any]] = []  # 打开着的listen流
        self.tools = {}
        self.server_info = {"name": "streamable-http-example-server", "version": "0.1.0"}

    def register_tool(self, name: str, handler, description: str = "", input_schema: Dict = None):
        """注册工具"""
        self.tools[name] = {
            "handler": handler,
            "description": description,
            "inputSchema": input_schema or {"type": "object", "properties": {}},
        }

    def _check_security(self, request) -> Optional[web.Response]:
        """校验Origin和认证；本地开发也不应默认暴露到0.0.0.0。"""
        origin = request.headers.get("Origin")
        if origin and origin not in self.allowed_origins:
            return web.Response(status=403, text="Origin not allowed")
        if self.bearer_token:
            expected = f"Bearer {self.bearer_token}"
            if request.headers.get("Authorization") != expected:
                return web.Response(status=401, text="Unauthorized")
        return None

    def _stamp(self, result: Dict[str, Any]) -> Dict[str, Any]:
        """在每个结果的 _meta 里回serverInfo：没有握手，身份只能靠它带回去。"""
        meta = dict(result.get("_meta") or {})
        meta.setdefault(SERVER_INFO_KEY, self.server_info)
        return {**result, "_meta": meta}

    def _error_response(
        self, request_id: Any, code: int, message: str, data: Any = None
    ) -> web.Response:
        """错误码决定HTTP状态，body仍是完整的JSON-RPC错误对象"""
        error: Dict[str, Any] = {"code": code, "message": message}
        if data is not None:
            error["data"] = data
        return web.json_response(
            {"jsonrpc": "2.0", "id": request_id, "error": error},
            status=ERROR_HTTP_STATUS.get(code, 200),
        )

    def _classify(self, method: str, params: Any, headers) -> Optional[tuple]:
        """信封与三个MCP头的校验阶梯，第一个不通过的就是结论。"""
        meta = params.get("_meta") if isinstance(params, dict) else None
        if not isinstance(meta, dict):
            return (
                INVALID_PARAMS,
                "params._meta must be an object carrying the required "
                f"{PROTOCOL_VERSION_KEY!r} and {CLIENT_CAPABILITIES_KEY!r} envelope keys",
                None,
            )
        missing = [k for k in (PROTOCOL_VERSION_KEY, CLIENT_CAPABILITIES_KEY) if k not in meta]
        if missing:
            return (
                INVALID_PARAMS,
                "params._meta is missing the required envelope key(s): " + ", ".join(missing),
                None,
            )
        version = meta[PROTOCOL_VERSION_KEY]
        # 头里的版本先决定走哪个时代的传输路径，所以它必须与信封里的版本一致
        if headers.get("MCP-Protocol-Version") != version:
            return (
                HEADER_MISMATCH,
                "MCP-Protocol-Version header does not match the envelope's protocol version",
                None,
            )
        if headers.get("Mcp-Method") != method:
            return (HEADER_MISMATCH, "Mcp-Method header does not match the body's method", None)
        name_key = NAME_BEARING_METHODS.get(method)
        if name_key is not None:
            body_value = params.get(name_key)
            if body_value is not None and headers.get("Mcp-Name") != body_value:
                return (
                    HEADER_MISMATCH,
                    f"Mcp-Name header does not match the body's {name_key!r} parameter",
                    None,
                )
        if version not in self.SUPPORTED_VERSIONS:
            return (
                UNSUPPORTED_PROTOCOL_VERSION,
                "Unsupported protocol version",
                {"supported": list(self.SUPPORTED_VERSIONS), "requested": version},
            )
        return None

    async def handle_mcp(self, request):
        """单端点只收POST；Mcp-Session-Id与Last-Event-ID一律忽略，本版已无会话与续传。"""
        if request.method != "POST":
            # GET流端点与DELETE终会话都已删除；规范建议回405并带上Allow
            return web.Response(status=405, headers={"Allow": "POST"})

        security_error = self._check_security(request)
        if security_error:
            return security_error

        accept = request.headers.get("Accept", "")
        if "application/json" not in accept or "text/event-stream" not in accept:
            return web.Response(status=406, text="Accept must include JSON and SSE")

        try:
            data = await request.json()
        except Exception:
            return self._error_response(None, PARSE_ERROR, "Parse error")

        # 本版HTTP传输上没有客户端到服务端的通知（取消靠关闭响应流），只收单个请求对象
        if not isinstance(data, dict) or data.get("id") is None or not isinstance(data.get("method"), str):
            return self._error_response(
                None, INVALID_REQUEST, "Body must be a single JSON-RPC request object"
            )

        request_id = data["id"]
        method = data["method"]
        params = data.get("params")
        rejection = self._classify(method, params, request.headers)
        if rejection is not None:
            return self._error_response(request_id, *rejection)

        if method == "subscriptions/listen":
            return await self._handle_listen(request, request_id, params)

        try:
            result = await self._process_request(method, params)
        except MCPRequestError as exc:
            return self._error_response(request_id, exc.code, exc.message, exc.data)
        return web.json_response(
            {"jsonrpc": "2.0", "id": request_id, "result": self._stamp(result)}
        )

    @staticmethod
    async def _write_event(resp: web.StreamResponse, message: Dict[str, Any]) -> None:
        """写一帧SSE"""
        body = json.dumps(message, ensure_ascii=False)
        await resp.write(f"event: message\r\ndata: {body}\r\n\r\n".encode())

    def _matches(self, subscriber: Dict[str, Any], method: str, params: Dict[str, Any]) -> bool:
        """只投递客户端确实订阅过的通知类型"""
        key = self.FILTER_KEYS.get(method)
        if key is not None:
            return subscriber["honored"].get(key) is True
        if method == "notifications/resources/updated":
            return (params or {}).get("uri") in subscriber["uris"]
        return False

    async def publish(self, method: str, params: Dict[str, Any] = None) -> None:
        """把一条变更通知投递给所有订阅了它的listen流"""
        for subscriber in list(self.subscribers):
            if self._matches(subscriber, method, params):
                subscriber["queue"].put_nowait(
                    {"jsonrpc": "2.0", "method": method, "params": dict(params or {})}
                )

    def close_subscriptions(self) -> None:
        """优雅关闭所有listen流，让客户端知道流是主动结束而不是掉线"""
        for subscriber in list(self.subscribers):
            subscriber["queue"].put_nowait(None)

    async def _handle_listen(self, request, request_id: Any, params: Dict[str, Any]):
        """subscriptions/listen的响应就是那条长连SSE，取代了已删除的GET流端点。"""
        requested = params.get("notifications") or {}
        honored = {key: True for key in self.FILTER_KEYS.values() if requested.get(key)}
        uris = list(requested.get("resourceSubscriptions") or [])
        if uris:
            honored["resourceSubscriptions"] = uris
        stream_meta = {SUBSCRIPTION_ID_KEY: request_id}

        resp = web.StreamResponse(
            status=200,
            headers={
                "Content-Type": "text/event-stream",
                "Cache-Control": "no-cache, no-transform",
                "X-Accel-Buffering": "no",
            },
        )
        await resp.prepare(request)

        subscriber = {"honored": honored, "uris": uris, "queue": asyncio.Queue()}
        self.subscribers.append(subscriber)
        try:
            # 确认帧必须是流上的第一帧，它回报服务端实际认下的订阅子集
            await self._write_event(resp, {
                "jsonrpc": "2.0",
                "method": "notifications/subscriptions/acknowledged",
                "params": {"notifications": honored, "_meta": stream_meta},
            })
            while True:
                try:
                    event = await asyncio.wait_for(subscriber["queue"].get(), timeout=15)
                except asyncio.TimeoutError:
                    await resp.write(b": ping\r\n\r\n")  # 保活，绕开代理的空闲超时
                    continue
                if event is None:
                    break
                event["params"] = {**event.get("params", {}), "_meta": stream_meta}
                await self._write_event(resp, event)
            # 优雅关闭：最后一帧是本请求的响应
            await self._write_event(resp, {
                "jsonrpc": "2.0",
                "id": request_id,
                "result": self._stamp({"resultType": "complete"}),
            })
        except ConnectionResetError:
            pass  # 客户端关掉了响应流，这就是取消信号，不必重连也不必补发
        finally:
            self.subscribers.remove(subscriber)
        return resp

    async def _process_request(self, method: str, params: Dict) -> Dict[str, Any]:
        """处理请求（具体业务逻辑）"""
        if method == "server/discover":
            # 取代initialize的能力探测：只读、可缓存、可被任意请求先行发起
            return {
                "resultType": "complete",
                "supportedVersions": list(self.SUPPORTED_VERSIONS),
                "capabilities": {"tools": {"listChanged": True}},
                "ttlMs": 60000,
                "cacheScope": "public",
            }
        elif method == "tools/list":
            return {
                "resultType": "complete",
                "ttlMs": 60000,
                "cacheScope": "public",
                "tools": [
                    {
                        "name": name,
                        "description": metadata["description"],
                        "inputSchema": metadata["inputSchema"],
                    }
                    for name, metadata in self.tools.items()
                ]
            }
        elif method == "tools/call":
            tool_name = params.get("name")
            arguments = params.get("arguments") or {}
            if tool_name not in self.tools:
                raise MCPRequestError(INVALID_PARAMS, f"Unknown tool: {tool_name}")
            result = await self.tools[tool_name]["handler"](arguments)
            return {"resultType": "complete", **result}
        # 未知方法是404，不是400：客户端据此判断对方根本没有这个能力
        raise MCPRequestError(METHOD_NOT_FOUND, f"Method not found: {method}")

    async def start(self):
        """启动服务器"""
        app = web.Application()
        # 单一端点，且只有POST是合法方法
        app.router.add_route("*", "/mcp", self.handle_mcp)

        runner = web.AppRunner(app)
        await runner.setup()
        site = web.TCPSite(runner, self.host, self.port)
        await site.start()
        print(f"Streamable HTTP MCP Server started at http://{self.host}:{self.port}/mcp")
        return runner

# Streamable HTTP传输使用示例
async def main():
    async with StreamableHttpMCPClient("http://127.0.0.1:8000") as client:
        # 没有握手，第一个请求就是探测：确认对方支持哪些协议版本
        discovered = await client.send_request("server/discover")
        print("Supported versions:", discovered["supportedVersions"])
        listing = await client.send_request("tools/list")
        print("Tools:", [tool["name"] for tool in listing["tools"]])
        print("Result:", await client.call_tool("add", {"a": 1, "b": 2}))