参数约束、结构化输出、错误处理、pytest 与 Streamable HTTP

这一课做完你会得到什么

一个「书店」MCP 服务器：参数有范围约束，返回值自带输出 Schema，查不到书时会给模型一条它能看懂、能据此纠正的错误。你还会为它写两条 pytest 测试，再把它改成 HTTP 服务，让 Claude Code 通过 URL 连接。

本课的 server.py 由官方 Tools、Structured Output、Handling errors 三页的示例片段合成，书目沿用官方示例里的三本书。

第 1 步：建项目并写书店服务器

uv init bookshop
cd bookshop
uv add "mcp[cli]"
uv add --dev pytest inline-snapshot

新建 server.py：

from typing import Annotated

from pydantic import BaseModel, Field

from mcp.server import MCPServer
from mcp.server.mcpserver.exceptions import ToolError
from mcp.types import ToolAnnotations

mcp = MCPServer("Bookshop")

CATALOG = {
    "Dune": ("Frank Herbert", 1965),
    "Neuromancer": ("William Gibson", 1984),
    "The Left Hand of Darkness": ("Ursula K. Le Guin", 1969),
}


class BookInfo(BaseModel):
    title: str
    author: str
    year: int = Field(description="Year of first publication.")


@mcp.tool(
    title="Search the catalog",
    annotations=ToolAnnotations(read_only_hint=True, open_world_hint=False),
)
def search_books(
    query: Annotated[str, Field(description="Title or author to search for.")],
    limit: Annotated[int, Field(ge=1, le=50, description="Max results.")] = 10,
) -> list[str]:
    """Search the catalog by title or author."""
    needle = query.lower()
    hits = [
        title
        for title, (author, _) in CATALOG.items()
        if needle in title.lower() or needle in author.lower()
    ]
    return hits[:limit]


@mcp.tool()
def get_book(title: str) -> BookInfo:
    """Look up one book in the catalog."""
    if title not in CATALOG:
        raise ToolError(f"No book titled {title!r} in the catalog.")
    author, year = CATALOG[title]
    return BookInfo(title=title, author=author, year=year)


if __name__ == "__main__":
    mcp.run()

这段代码里有四个新东西：

• Annotated[..., Field(...)]：给参数加说明和约束。ge=1, le=50 会写进输入 Schema，变成 "minimum": 1, "maximum": 50。写法和 FastAPI、Pydantic 完全一样。
• title= 与 ToolAnnotations：title 是给界面显示的名称。read_only_hint=True 表示这个工具不修改任何东西，open_world_hint=False 表示它只在一个封闭的数据集里查询。这些只是给客户端的提示，不是安全机制，不能指望客户端一定遵守。
• 返回类型 -> list[str] 和 -> BookInfo：返回类型注解就是输出 Schema，SDK 会据此生成 output_schema。
• ToolError：告诉模型「这次执行没成功」，错误信息会原样交给模型。

第 2 步：用内存客户端看约束、结构化输出和错误

新建 try_errors.py：

import anyio

from mcp import Client

from server import mcp


async def main() -> None:
    async with Client(mcp) as client:
        tools = await client.list_tools()
        for t in tools.tools:
            print(t.name, "->", t.output_schema.get("title"))

        r = await client.call_tool("search_books", {"query": "an"})
        print("1)", r.structured_content)

        r = await client.call_tool("search_books", {"query": "an", "limit": 999})
        msg = [ln for ln in r.content[0].text.splitlines() if "less than" in ln]
        print("2)", r.is_error, msg[0].split(" [")[0].strip())

        r = await client.call_tool("get_book", {"title": "Dune"})
        print("3)", r.structured_content)

        r = await client.call_tool("get_book", {"title": "Nothing"})
        print("4)", r.is_error, r.content[0].text)


anyio.run(main)

uv run python try_errors.py 2>/dev/null

对照结果（2026-10-25 实测；2>/dev/null 把服务器写到 stderr 的 INFO 日志藏起来）：

search_books -> search_booksOutput
get_book -> BookInfo
1) {'result': ['Dune', 'Neuromancer', 'The Left Hand of Darkness']}
2) True Input should be less than or equal to 50
3) {'title': 'Dune', 'author': 'Frank Herbert', 'year': 1965}
4) True Error executing tool get_book: No book titled 'Nothing' in the catalog.

逐行解读：

• 第 1、2 行：每个工具都带有输出 Schema。list[str] 不是 JSON 对象，所以 Schema 名为 search_booksOutput，结果被包进 {"result": [...]}。BookInfo 是 Pydantic 模型，Schema 直接用它的名字，结果就是模型的字段。
• 1)：搜 an 命中三本书，因为 Frank Herbert、Neuromancer、Ursula K. Le Guin 里都有 an。
• 2)：limit=999 超过了 le=50。SDK 在调用你的函数之前就拦下了这次请求，返回 is_error=True，而且错误文本模型能读懂，下一次会自动改用合法的值。
• 3)：structured_content 给程序用，content 里同时有一份 JSON 文本给模型看。
• 4)：ToolError 的信息加上了 Error executing tool get_book: 前缀，放在 content 里，模型可以据此换个书名重试。

官方特别提醒：不要用 return 返回错误信息。 返回的字符串 is_error 是 False，模型和客户端都会以为工具执行成功了。出错时要 raise。

第 3 步：分清三种失败

官方用一个问题来判断该抛哪种异常：换一个更聪明的模型，能不能避免这次失败？ 能，就抛 ToolError；不能，就抛 MCPError。其他没有预料到的异常都算崩溃。下面这段 errors_demo.py 可以直接运行，对比后两种情况：

import anyio

from mcp import Client, MCPError
from mcp.server import MCPServer
from mcp.types import INVALID_PARAMS

mcp = MCPServer("ErrorsDemo")
CATALOG = {"Dune": "Frank Herbert"}


@mcp.tool()
def strict_author(title: str) -> str:
    """Reject unknown titles at the protocol level."""
    if title not in CATALOG:
        raise MCPError(code=INVALID_PARAMS, message=f"No book titled {title!r}.")
    return CATALOG[title]


@mcp.tool()
def raw_author(title: str) -> str:
    """No check at all: a missing title raises KeyError."""
    return CATALOG[title]


async def main() -> None:
    async with Client(mcp) as client:
        try:
            await client.call_tool("strict_author", {"title": "Nothing"})
        except MCPError as e:
            print("MCPError:", e.error.code, e.error.message)

        r = await client.call_tool("raw_author", {"title": "Nothing"})
        print("崩溃:", r.is_error, r.content[0].text)


anyio.run(main)

uv run python errors_demo.py

对照结果（2026-10-25 实测，stdout 部分）：

MCPError: -32602 No book titled 'Nothing'.
崩溃: True Error executing tool raw_author

三种失败的区别：

-32602 就是 INVALID_PARAMS。崩溃时异常文本不会发给模型，因为它可能暴露服务器内部信息。完整的堆栈会以 ERROR 级别写进服务器日志：Tool 'raw_author' raised an unexpected exception。这个演示只是为了展示机制；对书名查不到这种情况，正确做法仍然是第 1 步的 ToolError。

第 4 步：写 pytest 测试

新建 test_server.py。raise_exceptions=True 只影响内存连接：服务器内部的协议层故障会直接抛到测试里，而工具里抛出的 ToolError 依然会变成 is_error=True 的结果，所以测试要对结果做断言：

import pytest
from inline_snapshot import snapshot

from mcp import Client

from server import mcp


@pytest.fixture
def anyio_backend():
    return "asyncio"


@pytest.fixture
async def client():
    async with Client(mcp, raise_exceptions=True) as c:
        yield c


@pytest.mark.anyio
async def test_get_book(client: Client):
    result = await client.call_tool("get_book", {"title": "Dune"})
    assert result.structured_content == snapshot(
        {"title": "Dune", "author": "Frank Herbert", "year": 1965}
    )


@pytest.mark.anyio
async def test_get_book_missing(client: Client):
    result = await client.call_tool("get_book", {"title": "Nothing"})
    assert result.is_error is True
    assert "No book titled 'Nothing'" in result.content[0].text

uv run pytest -q

对照结果（2026-10-25 实测）：

..                                                                       [100%]
2 passed in 0.57s

anyio_backend 夹具返回 "asyncio"，如果你用 trio 就改成 "trio"。snapshot(...) 来自 inline-snapshot：括号里留空运行 uv run pytest --inline-snapshot=create，它会把实际结果自动填进去。不想用它的话，改成普通的 == 比较也可以。每个测试拿到的都是一条新的内存连接，不用起进程，也不占端口。

第 5 步：改用 Streamable HTTP，并让 Claude Code 通过 URL 连接

stdio 只能在本机由宿主启动。想让别人或别的机器使用，就要提供一个 URL。新建 serve_http.py：

from server import mcp

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

uv run python serve_http.py

对照结果（2026-10-25 实测，节选）：

INFO:     Uvicorn running on http://127.0.0.1:3001 (Press CTRL+C to quit)

另开一个终端，新建 http_client.py 验证：

import anyio

from mcp import Client


async def main() -> None:
    async with Client("http://127.0.0.1:3001/mcp") as client:
        tools = await client.list_tools()
        print("工具：", [t.name for t in tools.tools])
        r = await client.call_tool("get_book", {"title": "Neuromancer"})
        print(r.structured_content)


if __name__ == "__main__":
    anyio.run(main)

uv run python http_client.py

对照结果（2026-10-25 实测）：

工具： ['search_books', 'get_book']
{'title': 'Neuromancer', 'author': 'William Gibson', 'year': 1984}

保持服务器运行，把它登记到 Claude Code：

claude mcp add --transport http bookshop http://127.0.0.1:3001/mcp
claude mcp list

对照结果（依据 Claude Code 官方文档）：先打印 Added ...，然后 claude mcp list 中 bookshop 显示 ✔ Connected。进入会话后，问「Who wrote Neuromancer?」，Claude Code 会调用 get_book。

关于 run() 的几个要点：

• 传输相关的参数都写在 run() 里，不能写进 MCPServer(...)。写错位置会直接报 TypeError: MCPServer.__init__() got an unexpected keyword argument 'port'。
• 默认监听 127.0.0.1:8000，路径是 /mcp；host、port、streamable_http_path 都可以改。请求体默认上限 4 MiB，超过会返回 HTTP 413。
• stateless_http=True 表示每个请求都新建传输、不跟踪会话；json_response=True 表示用单个 JSON 响应代替 SSE 流，但这样就不能在调用过程中回调客户端。
• 要部署到 localhost 以外的地方，还需要处理 DNS 重绑定防护（transport_security）和鉴权，详见官方 Deploy & scale 页：https://py.sdk.modelcontextprotocol.io/run/deploy/

实验结束后按 Ctrl-C 停止服务器。不再需要时，执行 claude mcp remove bookshop。

自查清单

• limit=999 返回 is_error=True 和 less than or equal to 50
• get_book 的 structured_content 是三个字段的字典
• 能说出 ToolError、MCPError 和普通异常分别让模型看到什么
• uv run pytest -q 显示 2 passed
• http_client.py 通过 3001 端口拿到了 Neuromancer 的信息