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

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

## 适合谁

- 已理解 LLM 工具调用，但还没有搭建过 MCP Server 的开发者；
- 正在多个 Agent 或模型之间重复维护工具定义的团队；
- 希望先用只读、低风险方式验证 MCP，而不是直接迁移生产工具的人。

## 先明确边界

**原文事实**：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. 创建目录

```bash
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` 导入路径，因此这里显式限制主版本：

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

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

### 2. 编写 Server

文件：`server.py`

```python
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()
```

这一步解决两个问题：

- `FastMCP` 创建标准 Server；
- `@mcp.tool()` 把普通 Python 函数注册为可发现工具，并根据类型标注生成参数 Schema。

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

### 3. 编写 Client

文件：`client.py`

```python
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. 运行并验收

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

期望输出中的关键两行：

```text
tools: ['add']
result: 12
```

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

**验收标准**：

- Client 能成功初始化；
- `list_tools()` 只返回 `add`；
- 参数 `7` 和 `5` 返回文本结果 `12`；
- Server 未访问网络，也未读取任何密钥。

### 5. 验证失败输入

把调用参数临时改成：

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

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

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

```python
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())
```

运行：

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

期望输出：

```text
checks: 4 passed
```

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

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

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

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

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

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

```bash
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 至少记录以下契约：

```yaml
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 使用：

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

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

```bash
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

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

- 身份认证和密钥轮换；
- TLS 与网络隔离；
- 会话复用和握手延迟；
- 并发限制、超时和熔断；
- 工具级授权，而不只是 Server 级授权；
- 审批结果和调用链的关联 ID。

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

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

## 上线前检查表

- [ ] 工具名称和描述没有歧义或重名；
- [ ] 输入 Schema 有类型、范围和长度约束；
- [ ] Resource 与 Tool 的读写边界清晰；
- [ ] 写入、删除、通知和部署需要显式审批；
- [ ] 工具输出按不可信输入处理，防范提示词注入；
- [ ] Server 来源、版本和发布者可验证，防范工具投毒；
- [ ] 日志包含工具名、脱敏参数、状态、错误、审批和关联 ID；
- [ ] stdio Server 不向 stdout 写普通日志；
- [ ] 远程 Server 已验证认证、隔离、会话复用和超时；
- [ ] 先从只读、低风险工具试点，并保留原路径作为回退。

## 常见误区

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

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

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

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

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

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

## 验证证据

- `static_publish_ok`：由发布步骤验证 Markdown、独立 HTML 和远端非空状态。
- `mock_code_run_ok`：本教程的本地无网络 MCP 闭环及自动边界检查均已执行；关键输出为 `tools: ['add']`、`result: 12` 和 `checks: 4 passed`。
- `real_backend_contract_ok`：不适用；教程不接入 LLM 或外部业务后端。
- `real_backend_smoke_ok`：未执行；教程刻意保持本地、无密钥、无外部 API。

## 来源与边界

- 原文：[MCP Explained: How Modern AI Agents Connect to the Real World](https://towardsdatascience.com/mcp-explained-how-modern-ai-agents-connect-to-the-real-world/)
- 作者：Maria Mouschoutzi
- 来源：Towards Data Science
- 日期：2026 年 7 月 28 日
- 版本边界：原文示例采用 MCP Python 1.x 风格 API；本文闭环以 `mcp 1.29.0` 实测。`mcp 2.0.0` 的导入路径与本文不兼容，未在本教程中迁移。
- 内容边界：Host、Client、Server、Tools、Resources、Prompts 和主要风险来自原文；权限矩阵、审计字段、验收标准与故障处理属于实践扩展。