第六篇中，我通过真实客户端与服务端跑通了 MCP 的发现和调用，也区分了协议版本、传输方式与模型工具调用。接下来要解决的是一个业务问题：工具已经被发现，怎样让模型使用它，同时保证每次执行都在允许的范围内？

我继续使用资料研究与报告生成案例。本篇建立三个工具：搜索本地文档、读取文档、保存报告。它们使用统一结构返回结果，Host 负责工具适配与执行策略，Server 负责最终文件操作和确认验证。

本篇采用 mcp==2.3.0，本地实际使用 2026-07-28 协议和 stdio 传输。资料核对与验证日期为 2026-11-06。

文内包含全部七个源码与依赖文件，另有 07_mcp_document_tools/ 运行副本。演示读取人工测试资料，不联网、不调用模型。模型接口桥接通过真实 OpenAI SDK 与本地模拟响应验证，不能据此声称模型已经完成研究任务。

一、先把工具的能力范围说清楚

我把工作区分成 sources/ 和 reports/。前者只读，后者只接受新报告。读取工具不能读取刚写出的报告，保存工具不能覆盖来源。这两个目录由可信启动配置确定，模型只能提供受限文件名。

工具	输入	输出	业务约束
search_documents	查询词、结果数量	文件名、命中片段、扫描情况	查询最多 80 字符，最多三个结果
read_document	文件名	文件名和 UTF-8 正文	单文件最多 24,000 字节
save_report	文件名、正文、确认凭据	文件名和写入字节数	内容受限，确认有效，拒绝覆盖

文件名只接受字母或数字开头、以 .md 结尾的简单名称，例如 notes.md。路径分隔符、绝对路径和 ../ 都不属于输入契约。报告最多 8,000 字符，同时检查 UTF-8 编码后的字节数；字符数与文件大小不能混为一谈。

搜索采用区分大小写的字面匹配，不递归，最多检查 64 个目录项，并通过受限读取函数打开每个候选文件。命中数量达到上限、扫描达到上限时返回 truncated=true；这个字段表示搜索提前停止，不能理解为已经找全所有资料。不可读、非法或过大的候选来源计入 skipped。

这些范围决定了本篇的实际能力：在受控的小型 Markdown 资料目录内检索。没有语义排序、网络抓取或全文索引，也不应把“没有命中”解释成“现实中没有这类证据”。

二、从模型建议到文件操作，中间有两层检查

Host 把 Server 上的三个名称加上可信配置前缀，成为 docs__search_documents 等模型工具名。以后接入另一台 Server 时，应分配另一个唯一前缀，并在总注册表检查冲突。不能让 Server 自己挑选前缀覆盖已有工具。

发现与授权是两件事。我只注册本篇白名单中的工具，额外发现的工具不会自动进入模型 Schema。readOnlyHint 等注解有助于描述行为，但来自服务端的注解不能证明权限，也不能替代人工确认。资料 1

本篇 Server 进程固定服务一个可信范围，DOCS_SCOPE 与 DOCS_ROOT 在启动时绑定。未来 FastAPI 根据鉴权结果选择对应资源范围，不能直接把用户填写的字符串当成已认证身份。当前实现不具备多租户 HTTP 鉴权。

三、输入 Schema 之外，还需要业务契约

MCP 的 inputSchema 描述调用参数，模型接口中的工具 Schema 描述模型可以提出什么参数。适配后两者很接近，但并不完全相同：模型不应看到 approval_token，这个字段只能由可信 Host 注入。

我使用 Pydantic 严格输入模型，拒绝额外字段与隐式类型转换。例如 JSON 的 true 不能冒充搜索数量 1。Server 也验证自己的输入，不能依赖 Host 已经检查过。

输出采用 ToolResult，由 SDK 生成 outputSchema 并返回 structuredContent。Host 同时检查 MCP 层和业务层：资料 1

结果	本篇处理方式
resultType 不是 complete	本篇不支持继续输入，转为需核查状态
isError=true	读取记为工具失败；写入保守记为结果未知
结构不符合输出 Schema	不进入模型正常观察结果
业务 status=ok	本次操作成功，可读取 data
业务 status=rejected	参数、确认或目标条件不允许执行
业务 status=failed	读取或搜索不可完成
业务 status=result_unknown	核查外部状态，不直接重试

complete 只表示协议请求已经完成，不表示文件已经成功保存。isError=false 同样不能覆盖业务结果中的拒绝或未知。

适配器对整个序列化结果设置 32,000 字节上限，而不仅检查正文。SDK 可以同时返回文本块和结构结果，正文在两处出现时，即使文件未超过读取上限，完整响应仍可能被 Host 拒绝。超大结果不直接塞入上下文；写入结果过大时仍可能已经产生文件，所以不能标记成“没有执行”。这个上限在 SDK 解码之后生效，不是传输层内存沙箱。处理不可信服务时，还需要更早的传输限制。

四、文件路径怎样真正受到约束

单纯检查输入不含 ../ 仍不够：sources/link.md 可以是指向工作区外的符号链接。

我使用 POSIX 的目录文件描述符。启动时以 O_DIRECTORY | O_NOFOLLOW 打开固定目录，后续通过 dir_fd 相对打开文件，并使用 O_NOFOLLOW 拒绝最终文件符号链接。读取后用 fstat() 检查实际打开对象是否是普通文件，再限制读取字节数。资料 2

这里没有“先检查路径，再按相同字符串重新打开”的两个独立步骤。持有目录描述符还能避免目录路径后来变化时，程序意外转向另一个同名目录。O_NONBLOCK 避免打开异常 FIFO 时一直等待，普通文件读取仍是同步文件操作。

这套实现面向 macOS/Linux 和由可信本地用户维护的目录。它不防御恶意本地用户替换可信祖先目录、制造硬链接、读取进程环境或同时修改来源。Python 目录校验也不是系统级权限沙箱；后续容器部署还需要设置文件系统挂载和运行身份。

三个工具使用同步函数，SDK 2.3.0 在工作线程中执行同步工具。Host 中的 Client 调用是异步的，演示中的终端输入使用 asyncio.to_thread()，避免等待确认时阻塞事件循环。目录资源在服务退出时关闭，Client 则在整个执行期间复用一个连接。

五、确认必须绑定本次具体操作

用户说“可以保存”，系统需要知道批准的是哪个文件、哪一份内容、哪个身份范围，以及确认何时失效。

本篇用 HMAC 签名生成短期凭据，包含可信范围、工具名、文件名、报告 SHA-256 与到期时间。Server 验证签名后，再比较本次参数与凭据中的操作。内容、身份或目标改变时，原凭据不能继续使用。

Host 与本地 stdio 子进程共享随机密钥。密钥不进入模型上下文；模型 Schema 删除 approval_token，本地输入模型也拒绝模型自行提交这个字段。模型调用 save_report 时，适配器先返回 waiting_approval 与候选内容，可信程序审核后才能调用 approved=True 的内部路径。

approved 不是模型参数，而是 Host 的可信控制输入。本篇终端演示展示完整候选报告，只有输入 approve 才继续。实际网页应把相同的候选内容展示给用户，不能审核一个摘要后悄悄保存另一份正文。

这还是同步教学确认：没有持久审批记录、一次性凭据消费表和跨进程恢复。签名证明内容绑定，不等于数据库已经记录了用户的批准；第十一篇会补上持久暂停、批准者验证与恢复。

保存通过 O_CREAT | O_EXCL 创建新文件，已有文件直接拒绝。它只解决同名覆盖问题，没有实现幂等：文件写到一半发生异常，目标可能已经存在；响应丢失后重试，也可能只得到“已存在”。我保留 result_unknown，把操作记录、结果查询与恢复交给第十二篇。

六、工具注册不是永远不变的快照

工具列表可能分页，也可能在任务运行期间变化。适配器最多读取四页、十六个工具，拒绝重复名称、重复游标和过大的工具描述。每次执行前绕过缓存重新发现，发现名称、参数、输出或描述变化时拒绝旧调用。

这样处理便于小工具集教学，代价是额外发现请求。生产环境可结合工具列表变化通知和带版本的注册表减少重复请求，但仍需要定义任务使用哪一版契约。检查与调用之间仍有时间窗口；严格版本绑定需要 Server 接受明确的契约版本或其他协调机制。

我只接受本篇使用的有限 JSON Schema 子集，并限制深度与节点数量，拒绝 $ref，避免校验时隐式解析外部 Schema。即便是有效 JSON Schema，也不代表任意模型供应商都支持其中的关键字。

模型接口使用 Chat Completions 工具格式和 strict=false，正确性依赖本地严格校验。ToolBridge.model_turn() 完成一轮真实 SDK 接口桥接：发送工具描述、解析工具调用、执行策略检查、追加匹配 tool_call_id 的结果消息。这个 ID 用于模型消息关联，与 MCP 请求 ID 分别管理。每次任务独立创建一个 Bridge，按顺序调用；它的预算计数没有实现并发协调，不能共享给多个任务并行执行。

接回第四篇状态图时，我让模型节点提出行动，让执行节点调用 invoke()，让观察节点消费状态。遇到 waiting_approval 就路由到等待确认，遇到 result_unknown 就停止自动写入并请求核查。确认状态必须由图外可信输入驱动，不能把等待结果再交给模型，让它自行批准。

本篇没有复制完整图，也没有把文档内容当作可信指令。第五篇的正文外存与版本引用仍可在观察节点使用：先保存有效工具结果，再把选出的证据片段加入上下文。保存教学报告成功，只能证明工具链闭环，不能证明引用与结论可信。

七、完整源码

把以下七个文件保存在同一目录。无需复制前篇工程即可运行演示。

07_mcp_document_tools/
├── contracts.py
├── workspace.py
├── approval.py
├── server.py
├── adapter.py
├── demo.py
└── requirements.txt

7.1 contracts.py：严格输入与统一结果

"""输入结构与统一业务结果，供 Host 和 Server 共用。"""

from typing import Annotated, Literal

from pydantic import BaseModel, ConfigDict, Field

Filename = Annotated[str, Field(pattern=r"^[A-Za-z0-9][A-Za-z0-9_-]{0,63}\.md$")]
Query = Annotated[str, Field(min_length=1, max_length=80)]
Content = Annotated[str, Field(min_length=1, max_length=8000)]
Token = Annotated[str, Field(min_length=1, max_length=2048)]
MaxResults = Annotated[int, Field(ge=1, le=3)]


class StrictInput(BaseModel):
    model_config = ConfigDict(extra="forbid", strict=True)


class SearchInput(StrictInput):
    query: Query
    max_results: MaxResults = 3


class ReadInput(StrictInput):
    filename: Filename


class SaveInput(ReadInput):
    content: Content


class AuthorizedSaveInput(SaveInput):
    approval_token: Token


class ToolResult(BaseModel):
    model_config = ConfigDict(extra="forbid", strict=True)
    status: Literal["ok", "rejected", "failed", "result_unknown"]
    message: str
    data: dict = Field(default_factory=dict)


INPUTS = {
    "search_documents": SearchInput,
    "read_document": ReadInput,
    "save_report": SaveInput,
}

7.2 workspace.py：受限文件操作

"""固定目录能力：仅支持 POSIX，教学目录必须由可信调用方准备。"""

import os
import stat
from pathlib import Path

from contracts import Content, Filename
from pydantic import TypeAdapter

MAX_BYTES = 24_000
MAX_ENTRIES = 64
filename_validator = TypeAdapter(Filename)
content_validator = TypeAdapter(Content)


class Workspace:
    def __init__(self, root: Path):
        # 根目录来自可信配置；持有 fd，避免随后通过路径再次寻找目录。
        self.sources_fd = self.reports_fd = -1
        root_fd = os.open(root, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
        try:
            flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW
            self.sources_fd = os.open("sources", flags, dir_fd=root_fd)
            self.reports_fd = os.open("reports", flags, dir_fd=root_fd)
        except BaseException:
            self.close()
            raise
        finally:
            os.close(root_fd)

    def close(self) -> None:
        for name in ("sources_fd", "reports_fd"):
            descriptor = getattr(self, name)
            if descriptor >= 0:
                os.close(descriptor)
                setattr(self, name, -1)

    def read(self, filename: str) -> str:
        filename_validator.validate_python(filename, strict=True)
        descriptor = os.open(
            filename, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK,
            dir_fd=self.sources_fd,
        )
        with os.fdopen(descriptor, "rb") as source:
            info = os.fstat(source.fileno())
            if not stat.S_ISREG(info.st_mode) or info.st_size > MAX_BYTES:
                raise ValueError("来源必须是大小受限的普通文件。")
            raw = source.read(MAX_BYTES + 1)
        if len(raw) > MAX_BYTES:
            raise ValueError("来源超过读取上限。")
        return raw.decode("utf-8")

    def search(self, query: str, max_results: int) -> dict:
        # 不递归；扫描和读取均有上限，无法完整扫描时显式标记。
        hits, skipped, examined = [], 0, 0
        truncated = False
        with os.scandir(self.sources_fd) as entries:
            for entry in entries:
                if examined == MAX_ENTRIES:
                    truncated = True
                    break
                examined += 1
                if not entry.name.endswith(".md"):
                    continue
                try:
                    text = self.read(entry.name)
                except (OSError, ValueError):
                    skipped += 1
                    continue
                position = text.find(query)
                if position >= 0:
                    hits.append({
                        "filename": entry.name,
                        "excerpt": text[max(0, position - 40):position + 160],
                    })
                    if len(hits) == max_results:
                        truncated = True
                        break
        return {"hits": hits, "skipped": skipped, "examined": examined,
                "truncated": truncated}

    def save(self, filename: str, content: str) -> dict:
        filename_validator.validate_python(filename, strict=True)
        content_validator.validate_python(content, strict=True)
        raw = content.encode("utf-8")
        if len(raw) > MAX_BYTES:
            raise ValueError("报告超过字节上限。")
        # O_EXCL 拒绝已有文件；写入后异常可能已留下部分文件，不能盲重试。
        descriptor = os.open(
            filename, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW,
            0o600, dir_fd=self.reports_fd,
        )
        with os.fdopen(descriptor, "wb") as target:
            target.write(raw)
            target.flush()
            os.fsync(target.fileno())
        return {"filename": filename, "bytes": len(raw)}

7.3 approval.py：内容绑定的确认凭据

"""教学用短期确认凭据；密钥和可信身份不进入模型上下文。"""

import base64
import hashlib
import hmac
import json
import time

from contracts import SaveInput


def operation(filename: str, content: str, scope: str) -> dict:
    SaveInput(filename=filename, content=content)
    return {"scope": scope, "tool": "save_report", "filename": filename,
            "sha256": hashlib.sha256(content.encode("utf-8")).hexdigest()}


def sign(secret: str, scope: str, arguments: dict, ttl: int = 60) -> str:
    payload = {**operation(**arguments, scope=scope), "expires": int(time.time()) + ttl}
    encoded = base64.urlsafe_b64encode(
        json.dumps(payload, sort_keys=True, separators=(",", ":")).encode()
    ).decode()
    signature = hmac.new(secret.encode(), encoded.encode(), hashlib.sha256).hexdigest()
    return f"{encoded}.{signature}"


def verify(secret: str, scope: str, filename: str, content: str, token: str) -> None:
    try:
        encoded, signature = token.split(".")
        expected = hmac.new(secret.encode(), encoded.encode(), hashlib.sha256).hexdigest()
        if not hmac.compare_digest(signature, expected):
            raise ValueError
        payload = json.loads(base64.b64decode(encoded, altchars=b"-_", validate=True))
        expires = payload.pop("expires")
        if type(expires) is not int or expires <= time.time():
            raise ValueError
        if payload != operation(filename, content, scope):
            raise ValueError
    except (ValueError, KeyError, TypeError):
        raise ValueError("确认凭据无效、过期或与本次操作不匹配。") from None

7.4 server.py：注册与执行三个工具

"""stdio MCP 服务；stdout 专用于协议帧。"""

import os
from pathlib import Path

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

from approval import verify
from contracts import (
    AuthorizedSaveInput, Content, Filename, MaxResults, Query, ReadInput,
    SearchInput, Token, ToolResult,
)
from workspace import Workspace

server = MCPServer("document-tools", version="1.0.0")
workspace: Workspace
secret: str
scope: str
readonly = ToolAnnotations(read_only_hint=True, open_world_hint=False)


@server.tool(annotations=readonly, structured_output=True)
def search_documents(query: Query, max_results: MaxResults = 3) -> ToolResult:
    """在允许的教学来源目录搜索文本，返回最多三个片段。"""
    try:
        args = SearchInput(query=query, max_results=max_results)
        data = workspace.search(args.query, args.max_results)
        return ToolResult(status="ok", message="搜索完成。", data=data)
    except (OSError, ValueError):
        return ToolResult(status="failed", message="搜索参数或来源不可用。")


@server.tool(annotations=readonly, structured_output=True)
def read_document(filename: Filename) -> ToolResult:
    """读取允许来源目录中的单个 Markdown 文件；不接受目录路径。"""
    try:
        ReadInput(filename=filename)
        return ToolResult(status="ok", message="读取完成。",
                          data={"filename": filename, "content": workspace.read(filename)})
    except (OSError, ValueError):
        return ToolResult(status="failed", message="来源不存在、不可读或超出限制。")


@server.tool(
    annotations=ToolAnnotations(read_only_hint=False, destructive_hint=False,
                                idempotent_hint=False, open_world_hint=False),
    structured_output=True,
)
def save_report(filename: Filename, content: Content, approval_token: Token) -> ToolResult:
    """凭 Host 确认凭据创建报告；拒绝覆盖已有文件。"""
    try:
        AuthorizedSaveInput(filename=filename, content=content, approval_token=approval_token)
        verify(secret, scope, filename, content, approval_token)
    except ValueError:
        return ToolResult(status="rejected", message="参数或确认凭据无效。")
    try:
        data = workspace.save(filename, content)
        return ToolResult(status="ok", message="报告已保存。", data=data)
    except FileExistsError:
        return ToolResult(status="rejected", message="同名报告已存在，未覆盖。")
    except ValueError:
        return ToolResult(status="rejected", message="报告超过保存限制。")
    except OSError:
        # 创建、写入、flush、fsync、close 都可能失败，不能推断没有副作用。
        return ToolResult(status="result_unknown", message="保存结果需核查，请勿直接重试。")


def main() -> None:
    global workspace, secret, scope
    secret, scope = os.environ["DOCS_SECRET"], os.environ["DOCS_SCOPE"]
    if len(secret) < 32 or not scope:
        raise ValueError("需要可信身份范围和至少 32 字符的随机密钥。")
    workspace = Workspace(Path(os.environ["DOCS_ROOT"]))
    try:
        server.run(transport="stdio")
    finally:
        workspace.close()


if __name__ == "__main__":
    main()

7.5 adapter.py：发现、模型适配与执行策略

"""有限工具适配；SDK 与模型供应商的格式在此处衔接。"""

import copy
import json

from jsonschema import Draft202012Validator, ValidationError
from mcp import Client

from approval import sign
from contracts import INPUTS, ToolResult

PREFIX = "docs__"
MAX_SCHEMA_BYTES = 12_000
MAX_OUTPUT_BYTES = 32_000


def json_size(value) -> int:
    return len(json.dumps(value, ensure_ascii=False, separators=(",", ":")).encode())


def check_schema(schema: dict) -> None:
    """仅接受本篇需要的有限子集；拒绝引用，避免隐式解析外部 Schema。"""
    keywords = {
        "type", "title", "description", "properties", "required", "additionalProperties",
        "items", "enum", "default", "pattern", "minLength", "maxLength",
        "minimum", "maximum", "anyOf",
    }
    pending, visited = [(schema, 0)], 0
    while pending:
        node, depth = pending.pop()
        visited += 1
        if visited > 128 or depth > 8 or not isinstance(node, (dict, bool)):
            raise ValueError("Schema 复杂度不受支持。")
        if isinstance(node, bool):
            continue
        if set(node) - keywords:
            raise ValueError("Schema 包含不受支持的关键字。")
        pending.extend((child, depth + 1) for child in node.get("properties", {}).values())
        pending.extend((child, depth + 1) for child in node.get("anyOf", []))
        for key in ("items", "additionalProperties"):
            if key in node:
                pending.append((node[key], depth + 1))
    Draft202012Validator.check_schema(schema)


class ToolBridge:
    def __init__(self, client: Client, secret: str, scope: str):
        self.client, self.secret, self.scope = client, secret, scope
        self.tools: dict = {}
        self.remaining_calls = 12

    async def discover(self) -> dict:
        found, cursor, seen_cursors = {}, None, set()
        for _ in range(4):
            page = await self.client.list_tools(cursor=cursor, cache_mode="bypass")
            if page.result_type != "complete":
                raise ValueError("不支持的工具列表响应。")
            for tool in page.tools:
                if tool.name in found:
                    raise ValueError("服务端工具名称重复。")
                found[tool.name] = tool
                if len(found) > 16:
                    raise ValueError("工具数量超限。")
            cursor = page.next_cursor
            if not cursor:
                break
            if cursor in seen_cursors:
                raise ValueError("工具分页游标重复。")
            seen_cursors.add(cursor)
        else:
            raise ValueError("工具分页超限。")
        allowed = {}
        for name in INPUTS:
            if name not in found:
                continue
            tool = found[name]
            if json_size(tool.model_dump(by_alias=True)) > MAX_SCHEMA_BYTES:
                raise ValueError("工具描述过大。")
            if tool.input_schema.get("type") != "object" or not tool.output_schema:
                raise ValueError("工具结构不受支持。")
            check_schema(tool.input_schema)
            check_schema(tool.output_schema)
            allowed[PREFIX + name] = tool
        return allowed

    async def refresh(self) -> None:
        self.tools = await self.discover()

    def model_tools(self) -> list[dict]:
        exposed = []
        for alias, tool in self.tools.items():
            schema = copy.deepcopy(tool.input_schema)
            if tool.name == "save_report":
                schema["properties"].pop("approval_token", None)
                schema["required"] = [
                    name for name in schema.get("required", []) if name != "approval_token"
                ]
            schema["additionalProperties"] = False
            exposed.append({"type": "function", "function": {
                "name": alias, "description": tool.description or tool.name,
                "parameters": schema, "strict": False,
            }})
        return exposed

    async def invoke(self, alias: str, arguments: dict, *, approved: bool = False) -> dict:
        tool = self.tools.get(alias)
        if tool is None:
            return {"status": "rejected", "message": "工具未获授权或尚未发现。"}
        try:
            # 本地契约拒绝模型携带凭据，也避免 bool 被当成整数。
            clean = INPUTS[tool.name].model_validate(arguments).model_dump()
        except ValueError:
            return {"status": "rejected", "message": "工具参数无效。"}
        if tool.name == "save_report" and not approved:
            return {"status": "waiting_approval", "message": "需审核文件名和完整报告内容。",
                    "candidate": clean}
        if self.remaining_calls <= 0:
            return {"status": "rejected", "message": "工具调用预算已耗尽。"}
        try:
            latest = await self.discover()
            current = latest.get(alias)
            if current is None or current.model_dump() != tool.model_dump():
                return {"status": "rejected", "message": "工具已移除或描述发生变化，请重新发现。"}
        except Exception:
            return {"status": "unavailable", "message": "工具发现失败，未发起业务调用。"}
        if tool.name == "save_report":
            clean["approval_token"] = sign(self.secret, self.scope, clean)
        try:
            Draft202012Validator(tool.input_schema).validate(clean)
        except ValidationError:
            return {"status": "rejected", "message": "参数不符合服务端 Schema。"}
        self.remaining_calls -= 1
        uncertain = "result_unknown" if tool.name == "save_report" else "unavailable"
        try:
            result = await self.client.call_tool(tool.name, clean, read_timeout_seconds=5)
            if result.result_type != "complete":
                return {"status": uncertain, "message": "不支持本次结果类型，需核查。"}
            if result.is_error:
                status = "result_unknown" if tool.name == "save_report" else "failed"
                return {"status": status, "message": "MCP 返回工具执行错误。"}
            if json_size(result.model_dump(by_alias=True)) > MAX_OUTPUT_BYTES:
                return {"status": uncertain, "message": "工具输出超过 Host 上限。"}
            Draft202012Validator(tool.output_schema).validate(result.structured_content)
            return ToolResult.model_validate(result.structured_content).model_dump()
        except Exception:
            # 不向模型透出异常正文，避免路径或确认凭据进入消息。
            return {"status": uncertain, "message": "调用或结果校验失败，需按状态处理。"}

    async def model_turn(self, model_client, model: str, messages: list[dict]) -> list[dict]:
        """真实 Chat Completions 接口的一轮桥接；批准只能由可信调用方传入。"""
        response = await model_client.chat.completions.create(
            model=model, messages=messages, tools=self.model_tools(),
            parallel_tool_calls=False,
        )
        message = response.choices[0].message
        updated = [*messages, message.model_dump(exclude_none=True)]
        for call in message.tool_calls or []:
            try:
                if len(call.function.arguments.encode()) > 32_000:
                    raise ValueError("模型参数过大。")
                arguments = json.loads(call.function.arguments)
                outcome = await self.invoke(call.function.name, arguments)
            except (ValueError, TypeError):
                outcome = {"status": "rejected", "message": "模型参数不是有效 JSON 对象。"}
            updated.append({"role": "tool", "tool_call_id": call.id,
                            "content": json.dumps(outcome, ensure_ascii=False)})
        return updated

7.6 demo.py：真实协议调用与终端确认

"""真实 stdio 调用 + 人工测试资料；不调用在线模型。"""

import asyncio
import json
import os
import secrets
import sys
from pathlib import Path
from tempfile import TemporaryDirectory

from mcp import Client, StdioServerParameters

from adapter import ToolBridge


async def run(root: Path, secret: str, scope: str) -> None:
    parameters = StdioServerParameters(
        command=sys.executable, args=[str(Path(__file__).with_name("server.py"))],
        env={**os.environ, "DOCS_ROOT": str(root), "DOCS_SECRET": secret, "DOCS_SCOPE": scope},
    )
    async with Client(parameters, read_timeout_seconds=5) as client:
        bridge = ToolBridge(client, secret, scope)
        await bridge.refresh()
        print("可用工具：", ", ".join(item["function"]["name"] for item in bridge.model_tools()))
        for alias, args in [
            ("docs__search_documents", {"query": "证据", "max_results": 2}),
            ("docs__read_document", {"filename": "notes.md"}),
        ]:
            print(json.dumps(await bridge.invoke(alias, args), ensure_ascii=False))
        candidate = {"filename": "report.md",
                     "content": "# 教学报告\n\n仅验证工具链，不能作为研究结论。\n"}
        print(json.dumps(await bridge.invoke("docs__save_report", candidate), ensure_ascii=False))
        print("待确认文件：", candidate["filename"])
        print(candidate["content"])
        answer = await asyncio.to_thread(input, "审核以上完整内容后，输入 approve 保存：")
        if answer.strip() != "approve":
            print("已取消保存。")
            return
        outcome = await bridge.invoke("docs__save_report", candidate, approved=True)
        print(json.dumps(outcome, ensure_ascii=False))
        if outcome["status"] == "ok":
            print("读取临时产物核对：", (root / "reports" / "report.md").read_text())


def main() -> None:
    with TemporaryDirectory(prefix="mcp-docs-") as directory:
        root = Path(directory)
        (root / "sources").mkdir()
        (root / "reports").mkdir()
        (root / "sources" / "notes.md").write_text(
            "# 人工测试资料\n\n证据正文存放在外部，图状态保存版本引用。\n", encoding="utf-8",
        )
        asyncio.run(run(root, secrets.token_hex(32), "teaching/user-1"))
    print("演示临时工作区已清理。")


if __name__ == "__main__":
    main()

7.7 requirements.txt：本篇直接依赖

mcp==2.3.0
pydantic==2.14.3
jsonschema==4.26.0
openai==2.6.1

OpenAI 是可选模型桥接所需依赖，demo.py 不发模型请求。MCP 2.x 与 OpenAI SDK 在本次环境分别使用 HTTPX2 2.16.0 与 HTTPX 0.28.1，它们是不同包，不能因为名称接近就互换客户端。资料 3

八、运行和观察结果

本地验证使用 Python 3.13.13。代码使用 Python 3.11 以上可用的语法和 API，本篇未验证其他 Python 版本，也未验证 Windows。直接依赖固定了版本，传递依赖尚未生成完整锁文件。

在保存上述文件的目录运行：

python3 -m venv .venv
source .venv/bin/activate
python -m pip install -r requirements.txt
python demo.py

程序依次输出可用工具、搜索结果、读取结果和待确认报告。审核文件名与完整正文后输入 approve，保存工具返回 status=ok、文件名与字节数；输入其他内容则取消。

整个演示使用临时目录，退出时删除测试资料与报告，便于反复运行。终端直接读取产物只是 Host 的本地验收步骤，不表示 MCP 读取工具获得了报告目录权限。

如果独立启动 server.py，需要提前准备根目录及两个子目录，并提供 DOCS_ROOT、DOCS_SCOPE、DOCS_SECRET。这些来自可信启动配置；随机密钥至少 32 字符，与 Host 一致。不要把 Server 的 stdout 当作日志输出通道。

模型桥接接收已有 AsyncOpenAI 客户端、模型名与消息列表。联网之前应确认所选模型接受工具格式、设置模型请求超时与重试策略，并继续沿用第四篇总任务预算。本篇适配器最多发起十二次业务工具调用，每次读取超时为五秒；发现和模型调用不计入这个计数，不能把它当成完整任务时间或成本预算。

九、验证了什么，还有哪些边界

本篇在独立环境通过 36 项针对性检查，并区分验证方法：

方法	已验证内容
真实 stdio Client/Server	三工具发现、搜索、读取、保存，以及本次协议为 2026-07-28
真实服务端拒绝	绝对与相对越界、符号链接、大文件、缺失来源、伪造或过期凭据
确认与产物核对	内容、目标或身份改变后拒绝；未批准无文件；批准后内容一致；已有文件不覆盖
Host 契约检查	严格整数、未知工具、模型注入凭据、描述变化、调用预算、外部 Schema 引用
明确标记的响应替身	超大输出、错误结构、MCP 错误及写入未知状态
本地 SDK 模拟传输	Chat Completions 序列化、调用 ID 回传、写入建议进入确认、请求不含密钥或凭据
生命周期与本地故障注入	真实连接关闭后发现失败、目录描述符释放、fsync 失败后文件可能存在

连接检查验证的是已关闭连接不可继续调用，并未模拟网络中途断线或远端写入响应丢失。后两种情况需要在可靠执行篇结合操作记录演练。

正文代码已从 Markdown 独立提取后复跑，七个文件与运行副本逐字一致，批准与拒绝两条演示路径及上述 36 项检查均通过。教学检查不能证明生产负载、多用户隔离、真实模型选择工具的稳定性或研究结论质量。

下一篇我会建立证据标识、引用位置和报告结构，让“生成了一个文件”进一步变成“交付了一份能检查来源与结论关系的报告”。

参考资料

1. MCP 2026-07-28 Tools：发现与调用、结构结果、输出 Schema、工具错误与行为注解。

2. Python os 文档：目录文件描述符、dir_fd、O_NOFOLLOW、fstat() 与文件创建标志。可用性需结合操作系统。

3. MCP Python SDK 2.x Migration：MCPServer、统一 Client、snake_case 字段及 HTTPX2 迁移。本篇接口另以安装的 2.3.0 版本实测核对。

4. OpenAI Function Calling：工具定义、模型提出调用与应用回传结果。具体模型和供应商的 Schema 支持需单独验证。