写第一个 MCP Server

第一个 MCP server 可以只回答一件事：给定商品编号和仓库，返回一条库存记录。输入、输出和失败条件都能提前写清，接入宿主之后也容易核对。本篇使用三条虚构记录，不访问文件、网络或真实账户。本文交付可独立运行的库存业务核心，以及固定 SDK 版本的 stdio 适配与验收脚本；后者仍需在你的环境执行，不能把离线测试通过当作服务已经接通。

本文于 2026 年 10 月 5 日核对官方发布记录。该次核对记录的 Python SDK 稳定版本为 2.3.0，发布于 10 月 2 日；示例固定这个版本，使用 from mcp.server import MCPServer。旧教程中的 mcp.server.fastmcp.FastMCP 属于 1.x，不能与 2.x 的导入和类型字段混用。官方 release、PyPI 版本页和迁移指南可交叉核对。

SDK 要求 Python 3.10 或更高；这份工程统一按 Python 3.12 编写，其中验收脚本使用 asyncio.timeout。本次实际运行的是 Python 3.12.14，24 项标准库离线测试全部通过。环境预装的 MCP 是 1.29.0，本次没有安装或运行 2.3.0，因此 SDK 集成、真实宿主和 HTTP/OAuth 均未实测。代码已按该次核对的官方 API 核对，后文提供执行步骤与预期结果。

先选一条能完成的路线

1. 离线业务核心：现在即可复现。按“先设计工具的输入与结果”“把业务核心留在标准库里”“stdio 的每一行都要保持干净”和“把验收分成能独立复现的几层”的文件标签，将 schema.fixture.json、core.py、frame_check.py、test_offline.py 保存到同一新目录。运行后文的 python3 -m unittest -v test_offline.py，完成标志是退出码 0、Ran 24 tests 与 OK；已覆盖 PEN-01 在 east 返回 12、零库存成功和缺失记录失败。这条路线不安装 MCP，也不启动 stdio 服务。
2. 固定 SDK 接入：在离线层之后单独验证。再保存 server.py、requirements.txt 和 smoke_sdk.py，按“用 SDK 加上一层薄适配”的虚拟环境命令安装固定依赖，依次运行进程内、--stdio、--stdio --legacy 三种验收。每次都要检查退出码 0，以及输出中的 ok=true、quantity=12、对应 transport 与 protocol_version；这些是待取得的结果，不是本文实测记录。全部通过后，才到“接入宿主后，再走一次真实查询”核对宿主的实际调用与 source=fictional-fixture-v1。

若只想练参数与失败边界，完成第一条即可。若目的是接入 Codex，第二条和宿主调用缺一不可。SDK 阶段若提示版本不符、导入失败或断言失败，就停在该层排错，保留完整命令与错误，不用前一层的 OK 替代它。

本文关注服务端实现。已有服务如何加入 Codex，见Codex 里接 MCP；事件和订阅单独见MCP Events。先把一次确定的查询做完整，再按业务需求加入资源、提示模板或更复杂的交互。

用户面对的是宿主，例如编辑器或桌面应用。宿主中的 MCP client 负责连接服务、发现工具和发起调用。server 公布工具名称、说明与 schema，并在收到调用时执行自己的代码。本例不需要模型 API key，也不会自行规划下一步；它接收 get_stock 请求后，只做固定查询。官方接入说明解释了这条进程边界。

图 1 / 概念示意

当前协议需要特别留意生命周期。2026-07-28 使用无握手的请求模型，每次请求必须携带协议版本与客户端能力元数据；clientInfo 是建议每次携带、可按配置省略的可选字段（2026 年 10 月 14 日核对请求字段规范）。server/discover 用于发现服务。2025-11-25 及更早版本先交换 initialize，再发送 notifications/initialized。Python SDK 2.x 为两类客户端提供兼容路径，业务函数不需要自己判断消息属于哪一代。协议变更说明。

图 2 / 概念示意

capabilities 描述双方能处理哪些协议功能，不能当作账户权限。写服务时只注册实际提供的能力，让 SDK 处理发现与旧版初始化。排错要保存客户端版本和最终使用的协议版本；只看到 initialize 缺失，不能断言现代连接失败。本文不手写握手、请求分派或订阅实现。

另一个需要留意的迁移点是 Roots：2026-07-28 规范已将它标记为弃用，建议新实现不要采用，已有实现改用工具参数、资源 URI 或服务端配置来传递目录和文件。现有部署应先核对宿主版本与实际使用的协议版本，再安排迁移并保留仍需支持的旧版兼容；Roots 能力声明与 roots/list 的交互方式在两代协议中也有差别，不能直接照搬旧版消息。Roots 本身只是相关目录和文件的提示，不提供访问控制，迁移后仍要由服务端独立检查权限与路径。

get_stock 的参数是 query 对象，包含 sku 和 warehouse。SKU 使用大写字母开头，长度为 1 到 16；仓库只能是 east 或 west，省略时使用 east。参数里不提供命令、文件路径或任意 URL。这样的限制与这个小工具的业务范围一致，也能让模型更准确地构造请求。

server.py 用 Pydantic 的 StockQuery 描述输入。strict=True 禁止把不合适的类型悄悄转换，extra=forbid 拒绝 query 内的多余字段。这里的禁止范围是嵌套的 query，不应扩大解释为所有 JSON-RPC 层级都会拒绝额外字段。SDK 根据函数签名和模型生成 schema；接口变更后，应该重新检查 tools/list 返回的真实定义。工具参数文档。

图 3 / 概念示意

工具说明会被宿主用于理解何时适合调用。名称应稳定而具体，描述写清查询对象、数据范围和是否产生副作用；不要在描述里夹带要求宿主绕过确认的指令。扩展工具时也别把几十项无关能力塞进一个 execute 参数，应该让每个入口有可单独审查的用途。

返回值 Stock 是 dataclass，包含商品、仓库、数量、单位和来源。SDK 根据返回注解生成输出 schema，并提供结构化数据与文本内容。线路上的字段名是 structuredContent、isError；Python v2 对象使用 structured_content、is_error。客户端读取结果时要按所属层使用字段，不能照着 JSON 名称访问 Python 属性。结构化输出说明。

下面的 fixture 是人工维护的语义契约，便于评审和回归；它不是抓取到的 tools/list 响应。实际 schema 可能带有 title、$defs 和 $ref。SDK 验收脚本会解析本例的本地引用，再对比关键约束，不做整段 JSON 字符串比较。默认值、必填字段和枚举都要进入回归检查。

{
  "fixture_note": "Handwritten semantic contract; not a tools/list capture",
  "tool_name": "get_stock",
  "input_contract": {
    "type": "object",
    "properties": {
      "query": {
        "type": "object",
        "additionalProperties": false,
        "properties": {
          "sku": {
            "type": "string", "minLength": 1, "maxLength": 16,
            "pattern": "^[A-Z][A-Z0-9-]{0,15}$"
          },
          "warehouse": {
            "type": "string", "enum": ["east", "west"], "default": "east"
          }
        },
        "required": ["sku"]
      }
    },
    "required": ["query"]
  },
  "valid_arguments": {"query": {"sku": "PEN-01"}},
  "expected_result": {
    "sku": "PEN-01", "warehouse": "east", "quantity": 12,
    "unit": "piece", "source": "fictional-fixture-v1"
  }
}

将下面文件放进一个新目录。core.py 不依赖 MCP；它校验原始参数，从只读映射中取出记录，并返回不可变结果。即使未来从命令行或后台任务调用同一函数，也会经过这些业务校验。映射只防止这份示例代码修改数据，不代表进程获得了操作系统级的只读保护。

"""Read-only fictional inventory. No SDK, files, network or credentials."""
from dataclasses import dataclass
from types import MappingProxyType
import re

SKU_PATTERN = r"[A-Z][A-Z0-9-]{0,15}"
WAREHOUSES = ("east", "west")
_STOCK = MappingProxyType({
    ("east", "PEN-01"): 12,
    ("west", "PEN-01"): 4,
    ("east", "BOOK-01"): 0,
})


class QueryError(ValueError):
    """Expected input or lookup failure, safe to show to the caller."""


@dataclass(frozen=True)
class Stock:
    sku: str
    warehouse: str
    quantity: int
    unit: str = "piece"
    source: str = "fictional-fixture-v1"


def lookup_stock(sku: str, warehouse: str = "east") -> Stock:
    if type(sku) is not str or re.fullmatch(SKU_PATTERN, sku) is None:
        raise QueryError("sku must match [A-Z][A-Z0-9-]{0,15}")
    if type(warehouse) is not str or warehouse not in WAREHOUSES:
        raise QueryError("warehouse must be east or west")
    key = (warehouse, sku)
    if key not in _STOCK:
        raise QueryError("No fixture record for this sku and warehouse")
    return Stock(sku=sku, warehouse=warehouse, quantity=_STOCK[key])

库存为零和没有记录必须分开。BOOK-01 在 east 的数量为 0，是成功查询；同一商品在 west 没有记录，抛出 QueryError。示例不会把缺失值补成零，否则调用方可能据此做出错误的补货判断。来源字段明确说明数据来自虚构夹具，不能拿来回答真实库存问题。

把数据访问层和协议层拆开还有一个实际好处：出现错误时，可以直接运行核心函数，判断问题是否已经存在于业务代码。本例结果量固定且很小。以后查询多条数据，要先规定分页、数量上限和返回字段，避免一次调用把整份数据库或长日志送入模型上下文。

核心使用 fullmatch 校验 SKU，所以末尾换行、空格、路径片段和小写编号都会被拒绝。这里没有自动清洗输入；如果实际产品希望接受小写，需要先定义归一化规则，再同步调整 schema、函数和测试。不要只在工具说明里说输入会被限制，却让函数接受更宽的范围。

同一目录中再加入 server.py 与 requirements.txt。适配层只负责声明工具、转换预期业务错误，并启动 stdio。主入口放在 __main__ 判断下，导入模块做测试时不会立刻阻塞在服务器循环。read_only_hint 表示这项查询不修改数据，open_world_hint=False 表示数据来自封闭的演示集合。

"""MCP Python SDK 2.3.0 adapter; run with the pinned requirements."""
import logging
import sys

from mcp.server import MCPServer
from mcp.server.mcpserver.exceptions import ToolError
from mcp.types import ToolAnnotations
from pydantic import BaseModel, ConfigDict, Field
from typing import Literal

from core import QueryError, Stock, lookup_stock


class StockQuery(BaseModel):
    model_config = ConfigDict(extra="forbid", strict=True)
    sku: str = Field(
        min_length=1, max_length=16,
        pattern=r"^[A-Z][A-Z0-9-]{0,15}$",
        description="Fictional SKU, for example PEN-01",
    )
    warehouse: Literal["east", "west"] = "east"


mcp = MCPServer("Read-only inventory", version="1.0.0")


@mcp.tool(
    annotations=ToolAnnotations(read_only_hint=True, open_world_hint=False),
)
def get_stock(query: StockQuery) -> Stock:
    """Read one fictional inventory record; never reserve or modify stock."""
    try:
        return lookup_stock(query.sku, query.warehouse)
    except QueryError as exc:
        raise ToolError(str(exc)) from exc


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO, stream=sys.stderr)
    mcp.run(transport="stdio")

mcp==2.3.0
pydantic==2.13.4

requirements.txt 固定了本文直接使用的 MCP 与 Pydantic 版本。它尚未固定所有传递依赖，所以不是完整锁文件。在干净虚拟环境完成安装与下面的 SDK 验收后，再保留该环境的完整依赖清单。团队项目可以使用已有的锁文件工具和校验流程；不要把本机另一个项目的 freeze 输出直接搬过来。Pydantic 版本已在官方包页面核对。

python3.12 -m venv .venv
.venv/bin/python -m pip install -r requirements.txt
.venv/bin/python -m pip check
.venv/bin/python smoke_sdk.py
.venv/bin/python smoke_sdk.py --stdio
.venv/bin/python smoke_sdk.py --stdio --legacy
.venv/bin/python -m pip freeze > requirements.lock.txt

Windows 可用 py -3.12 -m venv .venv 创建环境，再把命令中的 .venv/bin/python 换成 .venv\Scripts\python.exe。安装阶段才需要网络；这份库存查询没有网络操作。本文未执行这些安装命令，也没有提供伪造的依赖锁。部署前应在目标操作系统上重新完成安装与验收。

查询没有记录时，core 抛出的 QueryError 被适配层转换为 ToolError。SDK 将它放进工具结果，并把 isError 标为 true，调用方可以据此改正商品或仓库。直接 return 一段报错文字会形成普通成功返回，所以不要用一条看似错误的字符串代替失败标志。SDK 错误处理。

图 4 / 概念示意

错误应按实际收到的响应区分：收到 JSON-RPC error 时，Client 会抛出 MCPError；收到工具结果时，先检查 is_error。本文固定的 Python SDK 2.3.0 MCPServer 会把未知工具名返回为 is_error=True 的工具失败结果，不能用“未知工具必然抛 MCPError”验收这份实现。请求方法不存在等协议层问题，与 result 内的工具失败仍是不同层；输入值越界和业务错误也不应笼统写成 -32602。固定版本的错误映射与Client 结果说明可以对照阅读。

错误消息需要告诉调用方可修正的条件，而不是鼓励无休止重试。一个不存在的编号连续查询不会变成存在；同样参数得到同样错误时，应停下确认输入。数据库限流等暂时故障可以设计有限重试，但还要区分服务不可用、调用者无权访问以及业务记录缺失。

本例只捕获预期的 QueryError。真实数据库连接失败或程序异常应在服务端留下脱敏诊断，并对调用方给出有限信息。不要把堆栈、查询语句或连接串直接拼进 ToolError。宿主怎样把错误呈现给模型也有差异，验收时应同时看客户端收到的类型与用户最终看到的内容。

stdio 服务由客户端作为子进程启动。stdin 接收消息，stdout 返回消息，每条 JSON-RPC 消息由换行分隔。诊断写入 stderr；stderr 中出现 INFO 日志本身并不表示调用失败。JSON 文本中的换行需要编码为转义字符，不能用缩进格式把一条消息输出成多行。当前 stdio 规范。

图 5 / 概念示意

Python SDK v2 在服务运行期间隔离协议流，并把普通 stdout 输出转到 stderr，减少工具函数或子进程打印干扰。但启动脚本、模块导入或 SDK 启动前的输出仍需要检查。最稳妥的写法是从一开始就让日志使用 stderr，不写启动横幅，也不把密码和完整环境变量加入调试日志。v2 传输迁移说明。

启动链也属于排错范围。shell 包装脚本输出一行欢迎词，就可能在 SDK 接管之前污染协议；使用绝对解释器路径直接执行文件更容易复现。不要用合并 stdout 与 stderr 的重定向收集协议记录，两条流合在一起之后，很难辨认原始错误来自哪里。

下面的 frame_check.py 只供离线测试识别合成响应夹具中的明显污染，检查一行 JSON、响应 ID 和 result/error 二选一。它不实现 MCP 传输、不验证完整 JSON-RPC schema，也没有覆盖通知、取消或订阅。真实通信仍交给 SDK，避免维护一份不完整的协议实现。

"""Small fixture checker, NOT an MCP transport or full protocol validator."""
import json


def check_fixture_frame(line: bytes) -> dict:
    if not line.endswith(b"\n") or line.count(b"\n") != 1:
        raise ValueError("fixture needs exactly one terminating LF")
    message = json.loads(line.decode("utf-8"))
    if type(message) is not dict or message.get("jsonrpc") != "2.0":
        raise ValueError("fixture needs a JSON-RPC object")
    if ("result" in message) == ("error" in message):
        raise ValueError("fixture needs result XOR error")
    if "id" not in message:
        raise ValueError("response fixture needs id")
    return message

readOnlyHint 是服务方声明的提示。它不会限制进程能读哪些文件，也不会替数据库建立只读账号。这个例子之所以只读，是因为代码只查固定映射，没有写入函数，也没有任意 shell、路径或 URL 入口。换成真实数据源后，需要按调用者、租户、资源和动作逐次授权。

图 6 / 概念示意

查询权限同样可能泄露商业或个人信息。先决定返回哪些字段，再给服务进程最小需要的账户权限；日志只保留排错所需的请求标识、阶段和脱敏错误。客户提供的商品编号不能变成未绑定参数的 SQL，也不能变成文件名拼接。工具名称被隐藏，不能替代真正的调用授权。

接入数据库时，先用专门的测试库与只读账户跑同样的查询，再决定服务需要哪些表和列。用户传入的 warehouse 只表达查询条件，不能作为身份凭证；客户端声称自己属于哪个组织，也需要独立验证。这个静态演示没有多用户身份，因此不能据此推导出多租户隔离已经实现。

服务返回的文档和字段仍属于外部内容。若检索结果要求上传凭据或改变权限，宿主应按原任务与授权范围判断，不能把返回文本当成系统指令。MCP 规范要求服务端验证输入、控制访问并处理输出安全；实际实现还需要结合数据源做限流和审计。工具安全要求与安全实践可作为评审入口。

先运行不需要 SDK 的测试。将 test_offline.py、frame_check.py、core.py 与 schema.fixture.json 放在同一目录，执行下面命令。预期输出 Ran 24 tests 和 OK。测试覆盖库存正例、零值、缺失记录、类型和长度边界、末尾换行、重复查询、不变性、核心静默输出，以及合成帧中的横幅、多行 JSON 和结果冲突。

python3 -m unittest -v test_offline.py

图 7 / 概念示意

这 24 项没有导入 MCP，不能证明 SDK 生成的 schema 或错误映射正确。下一层使用官方 Client 的进程内连接：列出工具，对比 schema，调用正常输入、非法参数与未知工具。再加 --stdio 启动真实子进程，最后加 --legacy 检查旧协议路径。以下完整脚本已按官方测试文档和Client 文档核对，但本次未运行。

"""SDK smoke test supplied for the reader; not run by the article author."""
import argparse
import asyncio
from importlib.metadata import PackageNotFoundError, version
import json
from pathlib import Path
import sys

ROOT = Path(__file__).resolve().parent


def require_pins():
    for package, expected in (("mcp", "2.3.0"), ("pydantic", "2.13.4")):
        try:
            actual = version(package)
        except PackageNotFoundError:
            actual = "not installed"
        if actual != expected:
            raise SystemExit(f"Need {package}=={expected}; found {actual}")


def query_schema(schema):
    node = schema["properties"]["query"]
    if "$ref" in node:
        prefix = "#/$defs/"
        assert node["$ref"].startswith(prefix)
        node = schema["$defs"][node["$ref"][len(prefix):]]
    return node


async def main(args):
    if not __debug__:
        raise SystemExit(
            "Assertions are disabled; run without -O/-OO and unset PYTHONOPTIMIZE."
        )
    require_pins()
    from mcp import Client, StdioServerParameters
    from mcp.types import TextContent
    from server import mcp

    fixture = json.loads((ROOT / "schema.fixture.json").read_text())
    target = mcp
    if args.stdio:
        target = StdioServerParameters(
            command=sys.executable,
            args=[str(ROOT / "server.py")],
            env={},
        )
    mode = "legacy" if args.legacy else "auto"
    async with asyncio.timeout(15), Client(target, mode=mode) as client:
        listed = await client.list_tools()
        assert [tool.name for tool in listed.tools] == ["get_stock"]
        tool = listed.tools[0]
        assert tool.annotations.read_only_hint is True
        assert tool.annotations.open_world_hint is False
        expected = fixture["input_contract"]["properties"]["query"]
        actual = query_schema(tool.input_schema)
        assert actual["additionalProperties"] is False
        assert actual["required"] == expected["required"]
        for key, value in expected["properties"].items():
            for field, content in value.items():
                assert actual["properties"][key][field] == content
        assert tool.output_schema is not None
        good = await client.call_tool("get_stock", fixture["valid_arguments"])
        assert not good.is_error
        assert good.structured_content == fixture["expected_result"]
        bad_inputs = [
            {},
            {"query": {"sku": "PEN-01", "path": "/tmp/no-read"}},
            {"query": {"sku": "PEN-01", "warehouse": "north"}},
            {"query": {"sku": 12}},
            {"query": {"sku": "PEN-01\n"}},
            {"query": {"sku": "MISSING"}},
        ]
        for item in bad_inputs:
            result = await client.call_tool("get_stock", item)
            assert result.is_error, item
        # MCPServer 2.3.0 reports unknown tools as tool-error results.
        # Unexpected JSON-RPC errors still propagate and fail this smoke test.
        unknown = await client.call_tool("missing_tool", {})
        assert unknown.is_error is True
        assert unknown.structured_content is None
        assert any(
            isinstance(block, TextContent) and block.text == "Unknown tool: missing_tool"
            for block in unknown.content
        )
        print(json.dumps({
            "ok": True,
            "transport": "stdio" if args.stdio else "in-memory",
            "protocol_version": client.protocol_version,
            "quantity": good.structured_content["quantity"],
        }))


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--stdio", action="store_true")
    parser.add_argument("--legacy", action="store_true")
    asyncio.run(main(parser.parse_args()))

测试报告应保留命令、解释器版本、退出码和失败输出。语法解析只能说明文件能被 Python 解析；它不能证明导入存在、依赖兼容或客户端交互正确。本文为所有 Python 文件做过语法检查，但没有把这个结果计入 SDK 通过项。给读者的可复现命令也不等于作者已经执行过这些命令。

现代成功路径预期打印 ok 为 true、quantity 为 12、protocol_version 为 2026-07-28；legacy 路径预期为 2025-11-25。脚本对工具列表、schema 约束、结构化结果和错误类型都有断言，任何失败应先停下检查。不能只保留最后一行打印而忽略退出码，或为了得到绿色结果删除失败断言。

"""Run with Python 3.12+, standard library only."""
from contextlib import redirect_stderr, redirect_stdout
from dataclasses import FrozenInstanceError, asdict
from io import StringIO
from pathlib import Path
import json
import unittest

from core import QueryError, WAREHOUSES, lookup_stock
from frame_check import check_fixture_frame

ROOT = Path(__file__).resolve().parent
FIXTURE = json.loads((ROOT / "schema.fixture.json").read_text())


class CoreTests(unittest.TestCase):
    def test_east(self):
        self.assertEqual(lookup_stock("PEN-01").quantity, 12)

    def test_west(self):
        self.assertEqual(lookup_stock("PEN-01", "west").quantity, 4)

    def test_zero_is_success(self):
        self.assertEqual(lookup_stock("BOOK-01").quantity, 0)

    def test_missing_record(self):
        with self.assertRaisesRegex(QueryError, "No fixture record"):
            lookup_stock("BOOK-01", "west")

    def test_unknown_warehouse(self):
        with self.assertRaises(QueryError):
            lookup_stock("PEN-01", "north")

    def test_bad_sku_shapes(self):
        for value in (None, True, 12, [], {}, "", "pen-01", "../PEN"):
            with self.subTest(value=value), self.assertRaises(QueryError):
                lookup_stock(value)

    def test_sku_whitespace(self):
        for value in ("PEN-01\n", " PEN-01", "PEN-01 ", "PEN\t01"):
            with self.subTest(value=value), self.assertRaises(QueryError):
                lookup_stock(value)

    def test_length_boundary(self):
        with self.assertRaisesRegex(QueryError, "No fixture record"):
            lookup_stock("A" * 16)
        with self.assertRaisesRegex(QueryError, "sku must match"):
            lookup_stock("A" * 17)

    def test_invalid_warehouse_types(self):
        for value in (None, True, [], {}, 1):
            with self.subTest(value=value), self.assertRaises(QueryError):
                lookup_stock("PEN-01", value)

    def test_no_normalization(self):
        with self.assertRaises(QueryError):
            lookup_stock("PEN-01", "EAST")

    def test_result_immutable(self):
        with self.assertRaises(FrozenInstanceError):
            lookup_stock("PEN-01").quantity = 999

    def test_repeat_does_not_mutate_stock(self):
        a = asdict(lookup_stock("PEN-01"))
        a["quantity"] = 999
        self.assertEqual(lookup_stock("PEN-01").quantity, 12)

    def test_fixture_result(self):
        args = FIXTURE["valid_arguments"]["query"]
        self.assertEqual(asdict(lookup_stock(**args)), FIXTURE["expected_result"])

    def test_fixture_schema_contract(self):
        schema = FIXTURE["input_contract"]
        self.assertEqual(schema["required"], ["query"])
        query = schema["properties"]["query"]
        self.assertFalse(query["additionalProperties"])
        self.assertEqual(query["required"], ["sku"])
        props = query["properties"]
        self.assertEqual(props["warehouse"]["enum"], list(WAREHOUSES))
        self.assertEqual(props["warehouse"]["default"], "east")
        self.assertEqual(props["sku"]["maxLength"], 16)

    def test_query_extra_key_is_rejected_by_core_signature(self):
        with self.assertRaises(TypeError):
            lookup_stock("PEN-01", path="/tmp/anything")

    def test_core_is_silent(self):
        out, err = StringIO(), StringIO()
        with redirect_stdout(out), redirect_stderr(err):
            lookup_stock("PEN-01")
        self.assertEqual((out.getvalue(), err.getvalue()), ("", ""))


class FrameFixtureTests(unittest.TestCase):
    def test_valid_result(self):
        frame = b'{"jsonrpc":"2.0","id":1,"result":{}}\n'
        self.assertEqual(check_fixture_frame(frame)["id"], 1)

    def test_protocol_error(self):
        frame = b'{"jsonrpc":"2.0","id":2,"error":{"code":-32602}}\n'
        self.assertIn("error", check_fixture_frame(frame))

    def test_stdout_banner(self):
        with self.assertRaises(ValueError):
            check_fixture_frame(b"server started\n")

    def test_multiline_json(self):
        with self.assertRaises(ValueError):
            check_fixture_frame(b'{\n"jsonrpc":"2.0","id":1,"result":{}}\n')

    def test_embedded_newline_is_escaped(self):
        obj = {"jsonrpc": "2.0", "id": 1, "result": {"text": "a\nb"}}
        frame = (json.dumps(obj) + "\n").encode()
        self.assertEqual(check_fixture_frame(frame), obj)

    def test_missing_terminator(self):
        with self.assertRaises(ValueError):
            check_fixture_frame(b'{"jsonrpc":"2.0","id":1,"result":{}}')

    def test_response_needs_id(self):
        with self.assertRaises(ValueError):
            check_fixture_frame(b'{"jsonrpc":"2.0","result":{}}\n')

    def test_result_and_error_cannot_coexist(self):
        with self.assertRaises(ValueError):
            check_fixture_frame(
                b'{"jsonrpc":"2.0","id":1,"result":{},"error":{}}\n'
            )


if __name__ == "__main__":
    unittest.main()

验收脚本不能在优化模式下跳过检查

smoke_sdk.py 使用 assert 核对工具列表、schema 与结果。Python 的 -O、-OO，或生效的 PYTHONOPTIMIZE=1/2 会移除这些断言；所以脚本现在先检查 __debug__，在检查依赖、导入 SDK 和创建客户端之前，以退出码 1 停止，并将原因写入 stderr，不输出 ok=true。遇到这条提示，应去掉启动命令的优化选项并取消该进程的 PYTHONOPTIMIZE 设置，再重新运行验收，不要删除保护分支。参见 Python 优化选项文档。

2026 年 10 月 20 日的离线控制流复核使用标准库测试替身，不连接 SDK：旧脚本在 -O 下会把错误数量 999、错误只读注解、非法输入被成功接受、未知工具被成功接受这四类故障都打印为 ok=true；普通模式能拒绝它们。补丁保留普通模式的检查，并在优化模式一律停止。下面新增的 5 项测试覆盖 -O、-OO、两种环境变量设置（各含进程内、--stdio、--stdio --legacy 三种入口）和普通模式到达依赖检查的位置。它们不启动真实 MCP server，也不能充当 SDK、stdio 或宿主通过记录。

python3 -m unittest -v test_smoke_runtime.py
python3 -O -m unittest -v test_smoke_runtime.py

两条命令均预期显示 Ran 5 tests、OK，退出码为 0。外层 unittest 使用不受 -O 移除的 assertEqual 方法；被测子进程的优化模式则应退出 1，普通模式测试只走到人为设置的依赖检查停止点。这些预期已在 Python 3.12.14 复现，原 24 项离线测试也重新通过。实际 SDK 2.3.0 仍未安装或执行。

"""Offline runtime guard tests only; no real MCP SDK or server is started."""
import os
from pathlib import Path
import subprocess
import sys
import unittest

ROOT = Path(__file__).resolve().parent
MESSAGE = "Assertions are disabled; run without -O/-OO and unset PYTHONOPTIMIZE."


class SmokeRuntimeTests(unittest.TestCase):
    def run_python(self, args, optimize=None):
        env = os.environ.copy()
        env.pop("PYTHONOPTIMIZE", None)
        if optimize is not None:
            env["PYTHONOPTIMIZE"] = optimize
        return subprocess.run(
            [sys.executable, *args],
            cwd=ROOT,
            env=env,
            capture_output=True,
            text=True,
            timeout=10,
        )

    def check_rejected(self, flags=(), optimize=None):
        for mode in ([], ["--stdio"], ["--stdio", "--legacy"]):
            with self.subTest(flags=flags, optimize=optimize, mode=mode):
                result = self.run_python(
                    [*flags, "smoke_sdk.py", *mode], optimize=optimize
                )
                self.assertEqual(result.returncode, 1)
                self.assertEqual(result.stdout, "")
                self.assertEqual(result.stderr.strip(), MESSAGE)

    def test_dash_O(self):
        self.check_rejected(("-O",))

    def test_dash_OO(self):
        self.check_rejected(("-OO",))

    def test_environment_one(self):
        self.check_rejected(optimize="1")

    def test_environment_two(self):
        self.check_rejected(optimize="2")

    def test_normal_mode_reaches_pin_check_without_sdk_import(self):
        # Stop at the dependency boundary, before any real MCP import.
        code = (
            "import asyncio, runpy\n"
            "ns = runpy.run_path('smoke_sdk.py')\n"
            "def stop():\n"
            "    raise SystemExit('PIN_CHECK_REACHED')\n"
            "ns['main'].__globals__['require_pins'] = stop\n"
            "asyncio.run(ns['main'](None))\n"
        )
        result = self.run_python(["-c", code])
        self.assertEqual(result.returncode, 1)
        self.assertEqual(result.stdout, "")
        self.assertEqual(result.stderr.strip(), "PIN_CHECK_REACHED")


if __name__ == "__main__":
    unittest.main()

SDK 子进程测试通过之后，把虚拟环境 Python 的绝对路径作为 command，把 server.py 的绝对路径作为一个 args 元素交给宿主。路径如何写进配置由宿主决定。Codex 的配置入口、允许列表和确认策略见Codex MCP 接入篇；本文不重复客户端配置表。

先确认工具目录中只有 get_stock，再给出已知输入 query.sku=PEN-01、query.warehouse=east。核对实际工具调用记录与结构化结果中的 quantity=12 和 source=fictional-fixture-v1。只看到应用回复十二件，不能证明它调用了服务；必须能找到对应调用。最后再试没有记录的组合，确认宿主没有把失败解释成零库存。

遇到 Cannot import name MCPServer，先在宿主实际使用的解释器里检查 MCP 版本；终端与应用可能使用不同虚拟环境。出现 unexpected keyword argument 或 camelCase 属性错误，核对是否混入 1.x 示例。若服务立即退出，查看 stderr、绝对路径和导入错误；若进程安静等待，可能只是还没有客户端发来请求。

改变工作目录再启动一次也值得检查。核心与服务文件的导入跟随脚本所在目录，验收脚本用自身位置读取夹具，不依赖终端当前目录。若你新增配置文件，应该说明它的位置来源；从用户 HOME 随意猜文件，容易让测试环境和实际宿主读取不同内容。

工具目录存在但调用失败时，先保存原始输入与脱敏错误，再检查参数层和业务层。schema.fixture.json 与 tools/list 不一致，可能是签名改动，也可能只是 title 或引用结构不同，应比较实际约束。对本例这种确定的只读查询，可以用同一输入复现；扩展为写入动作后，超时不代表动作没有完成，重试策略需要另行设计。

把 transport 改为 streamable-http 可以改变通信方式，却没有完成面向公网的部署。真实服务还需要证书、端点路由、请求与响应大小限制、超时、限流、日志策略和身份授权。本文不提供一个带固定 token 字典的所谓安全服务器，也不声称本地测试已经覆盖远程风险。

在 MCP 的 HTTP 授权模型中，server 通常是资源服务器：发现信息说明它信任哪个授权服务器，调用请求携带访问令牌，服务端验证令牌适用于自己，并检查必要范围。登录、签发令牌、租户隔离和对象级权限需要正确接入已有身份系统。令牌不应被原样转发给无关的下游服务。当前授权规范。

验证远程服务时还要明确测试打到了哪一层。反向代理接受连接，不能证明 MCP 请求到达业务函数；完成 OAuth 登录，也不能证明令牌具有目标记录的读取权限。把网络可达、令牌校验、协议响应和业务授权分别记录，可以更快定位一次访问被拒绝的原因。

Python SDK 的 HTTP token verifier 不保护 stdio，也不保护进程内 Client(mcp) 测试。因此内存测试成功不能作为 OAuth 的证据。远程验收至少要覆盖无令牌、过期令牌、错误受众、范围不足和跨租户请求，并检查反向代理之后的实际行为。对这些控制没有验证结果时，应保持本地演示范围。SDK 授权边界明确说明了这一点。

代码、图示和测试夹具均为本文编写，库存记录完全虚构。本文使用官方 SDK 文档、版本发布页和 MCP 2026-07-28 规范作为技术依据。核心测试的实际结果、SDK 示例的预期结果和远程部署建议已分别标明。复制公开页面中的完整文件即可进行本地实验，不依赖一个尚未上传的下载地址。

升级 SDK 前先核对发布记录与迁移指南，然后按业务核心、SDK 进程内、stdio 子进程、真实宿主的顺序回归。2.3.0 已调整部分旧协议的空元数据发送和工具 schema 查找行为，保存精确版本能让后续排错有据可查。