从零搭建一个可验证的 Python MCP 工具：从协议连接到权限治理

本教程整理自 Maria Mouschoutzi 的文章 MCP Explained: How Modern AI Agents Connect to the Real World。原文解释了 MCP 的 Host、Client、Server 架构，并用 Python 天气工具展示动态发现与调用。本文在此基础上增加一个完全本地、无 API Key 的可运行示例，以及权限、可观测性和传输方式的实践检查。

适合谁

先明确边界

原文事实：MCP 通过 Host、Client、Server 三层结构，把模型与工具的点对点适配变成标准协议连接；Server 可以暴露 Tools、Resources 和 Prompts。

本文提炼：最小验证闭环应包含“启动 Server → 初始化 Client 会话 → 发现工具 → 调用工具 → 检查结果”。

实践扩展：本文把示例改成不访问网络的加法工具，并增加最小权限、审批和日志设计。它们是保守的工程建议，不是原文声称的唯一实现方式。

原文方法覆盖清单

原文方法 是否纳入 落点 若遗漏原因
用 FastMCP 创建 Server 是 工作流一 —
用 @mcp.tool() 注册函数并生成 Schema 是 工作流一 —
Client 建立 stdio 会话 是 工作流一 —
运行时 list_tools() 发现工具 是 工作流一 —
用 call_tool() 调用工具 是 工作流一 —
天气 API 示例 否 改用本地加法工具 避免网络和第三方 API 影响最小闭环验证
Tools、Resources、Prompts 分类 概念纳入 工作流二 代码闭环只实现 Tool；Resource 与 Prompt 未实现
权限与提示词注入风险 是 工作流二 —
MCP 生态时间线 否 来源与边界 与动手闭环无直接关系，分享文章已保留

工作流一：运行一个最小 MCP Server 和 Client

1. 创建目录

mkdir -p mcp-minimal
cd mcp-minimal
touch server.py client.py checks.py

本教程按原文使用的 MCP Python 1.x API 编写。实测时，mcp==2.0.0 已不再提供原文使用的 mcp.server.fastmcp 导入路径，因此这里显式限制主版本：

uv run --no-project --with 'mcp[cli]>=1.29,<2' python client.py

如果要把示例变成正式项目，可在项目自己的依赖文件中固定并审查版本；不要无条件复制这个范围到生产环境。

2. 编写 Server

文件：server.py

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("local-calculator")


@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two integers."""
    return a + b


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

这一步解决两个问题：

工具描述不是普通注释。模型可能依据名称、docstring 和 Schema 决定是否调用它，因此描述必须简洁、准确且避免与其他工具重叠。

3. 编写 Client

文件：client.py

import asyncio
import sys
from pathlib import Path

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from mcp.types import TextContent


async def main() -> None:
    server = Path(__file__).with_name("server.py")
    params = StdioServerParameters(command=sys.executable, args=[str(server)])

    async with stdio_client(params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            tools = await session.list_tools()
            print("tools:", [tool.name for tool in tools.tools])

            result = await session.call_tool("add", {"a": 7, "b": 5})
            if not result.content:
                raise ValueError("tool returned no content")
            first = result.content[0]
            if not isinstance(first, TextContent):
                raise TypeError(
                    f"unexpected result type: {type(first).__name__}"
                )
            if result.isError:
                raise RuntimeError(first.text)
            print("result:", first.text)


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

Client 完成四件事：启动本地 Server 进程、初始化会话、发现工具、调用工具。它不包含 LLM；这反而能先验证协议层是否正确，避免把模型路由问题与工具通信问题混在一起。

4. 运行并验收

uv run --no-project --with 'mcp[cli]>=1.29,<2' python client.py

期望输出中的关键两行：

tools: ['add']
result: 12

SDK 可能同时向 stderr 输出请求处理日志；验收时不要把日志顺序当作业务结果。

验收标准：

5. 验证失败输入

把调用参数临时改成：

result = await session.call_tool("add", {"a": "seven", "b": 5})

期望：SDK 或 Server 返回参数验证错误，result.isError 为真，而不是悄悄把字符串当整数处理。验证后恢复原代码。

为了把成功与失败边界变成可重复检查，新增文件：checks.py

import asyncio
import sys
from pathlib import Path

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from mcp.types import TextContent


def text_of(result: object) -> str:
    content = getattr(result, "content", None)
    if not content or not isinstance(content[0], TextContent):
        raise AssertionError("expected non-empty TextContent")
    return content[0].text


async def main() -> None:
    server = Path(__file__).with_name("server.py")
    params = StdioServerParameters(command=sys.executable, args=[str(server)])

    async with stdio_client(params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            tools = await session.list_tools()
            assert [tool.name for tool in tools.tools] == ["add"]

            ok = await session.call_tool("add", {"a": 7, "b": 5})
            assert not ok.isError
            assert text_of(ok) == "12"

            wrong_type = await session.call_tool(
                "add", {"a": "seven", "b": 5}
            )
            assert wrong_type.isError

            missing_arg = await session.call_tool("add", {"a": 1})
            assert missing_arg.isError

            unknown_tool = await session.call_tool(
                "addx", {"a": 1, "b": 2}
            )
            assert unknown_tool.isError
            assert "Unknown tool" in text_of(unknown_tool)

    print("checks: 4 passed")


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

运行：

uv run --no-project --with 'mcp[cli]>=1.29,<2' python checks.py

期望输出：

checks: 4 passed

这四项分别验证工具发现与成功调用、错误类型、缺少参数和未知工具名。SDK 仍可能向 stderr 打印请求日志。

如果出现下面的导入错误：

ModuleNotFoundError: No module named 'mcp.server.fastmcp'

先检查当前未限定环境会选到哪个版本：

uv run --no-project --with 'mcp[cli]' python -c \
  'import importlib.metadata; print(importlib.metadata.version("mcp"))'

这个不限定版本的命令只用于诊断依赖漂移。若它显示 2.x，说明最新 SDK 与本文代码的主版本不匹配；再用下面的命令确认本文 1.x 环境：

uv run --no-project --with 'mcp[cli]>=1.29,<2' python -c \
  'import importlib.metadata; print(importlib.metadata.version("mcp"))'

之后应选择“按 1.x 运行本文示例”或“依据 2.x 官方迁移文档改写”，不要通过猜测导入路径掩盖兼容问题。

工作流二：把工具分成只读、可写和高风险三档

协议接通后，不要立刻把所有内部 API 暴露为 Tool。先制作一张能力清单：

能力 建议形态 默认策略 是否需要人工确认
读取公开文档 Resource 或只读 Tool 允许，限制路径和大小 否
查询内部数据库 只读 Tool 参数化查询、限制数据范围、脱敏日志 视数据敏感度
创建草稿 Tool 写入候选区，不直接发布 是
发送邮件或通知 Tool 默认禁用，展示收件人与正文预览 是
删除、部署、改生产数据 高风险 Tool 最小权限、双重确认、可回滚 是

对每个 Tool 至少记录以下契约：

name: create_draft
owner: content-platform
risk: write
input_schema: title/body
approval: required
side_effect: creates_unpublished_draft
rollback: delete_draft_by_id
logs:
  - correlation_id
  - sanitized_arguments
  - approval_decision
  - result_status

验收标准：每个可写能力都有明确 owner、side effect、approval 和 rollback；只读能力不能借用可写接口实现。

失败处理：若无法说明一个 Tool 会修改什么状态、如何撤销，就不要接入 Agent。先退回只读查询或人工执行。

工作流三：选择 stdio 还是远程 HTTP

本地单用户场景：先用 stdio

适合桌面工具、编辑器扩展和本机进程。优点是部署简单、网络暴露面小。

关键纪律：stdout 只承载协议帧，普通日志写入 stderr。随手 print() 到 stdout 可能破坏协议通信。

先做语法检查；这条命令不会检查 stdout 使用：

uv run --no-project --with 'mcp[cli]>=1.29,<2' \
  python -m compileall server.py client.py checks.py

再用 AST 扫描 Server 中直接调用的 print()；本例期望输出 print lines: []：

uv run --no-project --with 'mcp[cli]>=1.29,<2' python -c \
  'import ast,pathlib; t=ast.parse(pathlib.Path("server.py").read_text()); lines=[n.lineno for n in ast.walk(t) if isinstance(n,ast.Call) and isinstance(n.func,ast.Name) and n.func.id=="print"]; print("print lines:",lines); raise SystemExit(bool(lines))'

这只能发现直接 print()，不能证明所有依赖都遵守 stdio 纪律；仍需审查日志配置。需要日志时使用 Python logging 并显式指向 stderr。

多用户或跨主机场景：评估远程 HTTP

不要只看“能不能连通”，还要验证：

验收标准：在接入真实 Agent 前，用固定测试 Client 完成一次工具发现、一次成功调用、一次无权限调用和一次超时调用，并能从日志中还原完整链路。

失败处理：若无法区分协议错误、工具错误和权限拒绝，先补齐错误分类与追踪，不要扩大接入范围。

上线前检查表

常见误区

“采用 MCP 后，工具就自动安全了”

错误。MCP 标准化连接，但不会修复工具自身的权限、数据质量或副作用设计。

“所有 API 都应该迁移成 MCP”

没有必要。仍然有效的内部注册表或定制连接可以继续使用；只有重复适配和跨 Host 复用带来明确收益时，迁移才值得。

“Server 返回的内容可以直接交给模型”

不应默认信任。工具输出可能包含恶意指令或污染上下文，应限制来源、做结构校验，并把数据与指令边界分开。

验证证据

来源与边界