Agent AtlasAI Agent 知识库
06.03教程 · Tutorials

写一个 MCP Server 和 Client

从 stdio 到 Streamable HTTP，实现 Tools、Resources、Prompts，并用 Inspector 调试与审计。

11 分钟 进阶 更新于 2026 年 10 月 31 日
#教程#MCP

MCP 的价值是「写一次，多个客户端都能用」。这个教程从零写一个可用的文档 Server，再写 Client 调它，最后用 Inspector 和测试验证。

最终效果

text
acme-mcp/
├─ server.py          # MCP Server
├─ client.py          # 最小 Client
├─ docs/              # 本地文档
├─ tests/test_server.py
└─ requirements.txt

能力：

  • search_docs：检索文档；
  • read_doc：按 ID 读全文；
  • docs://{doc_id}：Resource；
  • summarize_doc：Prompt 模板。

前置

bash
pip install "mcp[cli]" pytest

Python 3.10+。如果你用 TypeScript，官方 SDK 也有对应实现，概念完全一致。

Step 1：准备数据

docs/ 放几个 Markdown 文件：

text
docs/refund.md
docs/shipping.md
docs/warranty.md

server.py 先加载它们：

python
from pathlib import Path

DOCS = {p.stem: p.read_text(encoding="utf-8") for p in Path("docs").glob("*.md")}

Step 2：写 Server

python
from __future__ import annotations

from pathlib import Path

from mcp.server.fastmcp import FastMCP

DOCS = {p.stem: p.read_text(encoding="utf-8") for p in Path("docs").glob("*.md")}
mcp = FastMCP("acme-docs")


@mcp.tool()
def search_docs(query: str, limit: int = 5) -> list[dict]:
    """检索 ACME 内部文档。

    何时使用：用户询问退款、配送、保修等政策，且需要引用原文。
    何时不要使用：需要订单实时状态时，用订单系统工具，不要用文档猜。

    Args:
        query: 关键词，尽量简短，例如 "退款 时效"。
        limit: 返回条数，1-10。
    """
    q = query.lower().strip()
    hits = []
    for doc_id, text in DOCS.items():
        if q in doc_id.lower() or q in text.lower():
            hits.append({"id": doc_id, "title": doc_id, "excerpt": text[:300]})
    return hits[: max(1, min(limit, 10))]


@mcp.tool()
def read_doc(doc_id: str) -> dict:
    """按 ID 读取文档全文。先用 search_docs 拿到 ID，不要猜。"""
    if doc_id not in DOCS:
        return {
            "ok": False,
            "error": {
                "code": "DOC_NOT_FOUND",
                "message": f"未知文档 {doc_id}，可用：{list(DOCS)}",
                "retryable": False,
            },
        }
    return {"ok": True, "id": doc_id, "text": DOCS[doc_id]}


@mcp.resource("docs://{doc_id}")
def doc_resource(doc_id: str) -> str:
    """把文档暴露为可读取的 Resource。"""
    if doc_id not in DOCS:
        raise ValueError(f"未知文档：{doc_id}")
    return DOCS[doc_id]


@mcp.prompt()
def summarize_doc(doc_id: str) -> str:
    """生成文档摘要的提示模板。"""
    return f"""请阅读文档 {doc_id}，输出：
1. 三条核心结论
2. 适用条件
3. 三个常见误解
每条不超过 40 字。"""


if __name__ == "__main__":
    mcp.run()  # 默认 stdio

三个原语的区别：

原语谁控制用途
Tool模型执行动作，可能写数据
Resource应用只读数据，按 URI 取
Prompt用户可复用模板，用户主动选择

Step 3：本地调试

MCP Inspector 是最快的调试方式：

bash
npx @modelcontextprotocol/inspector python server.py

打开它给出的地址，可以：

  • 看到 tools/list 返回的 Schema；
  • 直接调用工具，填参数；
  • 读取 Resource；
  • 查看每次 JSON-RPC 请求与响应。

先在这里确认工具能被发现、参数正确、错误可读，再接客户端。

Step 4：写 Client

client.py：

python
from __future__ import annotations

import asyncio
import json
import sys

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


async def main(query: str) -> None:
    params = StdioServerParameters(
        command=sys.executable,
        args=["server.py"],
    )
    async with stdio_client(params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            tools = await session.list_tools()
            print("工具：", [t.name for t in tools.tools])

            result = await session.call_tool("search_docs", {"query": query})
            for item in result.content:
                if item.type == "text":
                    print(item.text)

            result = await session.call_tool("read_doc", {"doc_id": "refund"})
            for item in result.content:
                if item.type == "text":
                    print(item.text[:500])


if __name__ == "__main__":
    asyncio.run(main(sys.argv[1] if len(sys.argv) > 1 else "退款"))

运行：

bash
python client.py "退款"

预期：

text
工具： ['search_docs', 'read_doc']
[{"id": "refund", "title": "refund", "excerpt": "# 退款政策..."}]
{"ok": true, "id": "refund", "text": "..."}

Step 5：暴露为远程 HTTP 服务

本地工具用 stdio；团队共享或云端部署用 Streamable HTTP：

python
if __name__ == "__main__":
    mcp.settings.host = "0.0.0.0"
    mcp.settings.port = 8000
    mcp.run(transport="streamable-http")

启动：

bash
python server.py
# MCP endpoint: http://localhost:8000/mcp

远程服务必须加认证。不要在公网暴露一个可以读文件、跑命令的 Server。最小要求：

  • OAuth 2.1 或短期 token；
  • 每个用户独立 scope；
  • 文件访问限制在固定目录；
  • 所有调用写审计日志；
  • 限制请求频率与返回大小。

Step 6：测试

tests/test_server.py：

python
import pytest

from server import DOCS, read_doc, search_docs


def test_search_returns_hits():
    hits = search_docs("退款", limit=3)
    assert hits
    assert len(hits) <= 3
    assert "id" in hits[0]


def test_read_unknown_doc_is_structured_error():
    result = read_doc("nope")
    assert result["ok"] is False
    assert result["error"]["retryable"] is False


def test_docs_loaded():
    assert "refund" in DOCS

运行：

bash
pytest -q

再补一类「协议级」测试：启动 Client，list_tools() 是否包含预期工具，call_tool 返回是否符合 Schema。工具改名、参数变更时，这类测试会第一时间报警。

设计要点

工具描述要写「何时不用」

模型靠描述选工具。只写「检索文档」不够，要写清楚：

  • 什么时候用；
  • 什么时候应该用别的工具；
  • 参数含义与单位；
  • 返回结构；
  • 是否有副作用。

返回要小

MCP 返回会进入模型上下文。read_doc 返回整篇 20 页文档，很容易撑爆。生产做法：

  • read_doc(doc_id, offset, limit) 分页；
  • 返回 summary + text；
  • 超过阈值时标注 truncated: true；
  • 列表工具限制 limit。

错误要可操作

json
{
  "ok": false,
  "error": {
    "code": "DOC_NOT_FOUND",
    "message": "未知文档，可用：refund, shipping, warranty",
    "retryable": false
  }
}

模型看到可用列表，就能换一个 ID；只返回「失败」它只能猜。

安全清单

  • 只暴露必要工具，不暴露 shell/SQL
  • 文件访问限制在 roots 目录
  • 远程服务使用 OAuth，scope 最小
  • 第三方 Server 锁定版本与哈希
  • 记录工具调用审计
  • 返回结果裁剪与分页
  • 不把 MCP 当安全边界
工具描述也可能被投毒

恶意 Server 可以在工具描述里写「忽略之前的指令」。接入第三方 Server 前要审查描述，并限制它能看到的数据。

部署与运维

  • stdio：随 Host 启动，适合本地工具；日志写 stderr，不要污染 stdout。
  • HTTP：可多客户端共享；需要反向代理、TLS、认证、限流。
  • 版本：Server 和 Client 都要固定版本；工具 Schema 变更视为破坏性变更。
  • 监控：记录调用次数、错误率、P95 延迟、返回大小。

验收清单

  • Inspector 能看到三个原语
  • Client 能列出并调用工具
  • Resource 能按 URI 读取
  • 未知文档返回结构化错误
  • pytest 通过
  • HTTP 模式有认证
  • 返回有分页/截断

扩展任务

  1. 加 create_ticket 写工具，要求幂等键；
  2. 把文档检索换成向量检索；
  3. 加 OAuth，按用户过滤可读文档；
  4. 用 TypeScript SDK 重写一遍，比较开发体验；
  5. 接进一个支持 MCP 的客户端，观察模型如何选择工具。

小结

  • MCP 用统一协议把工具、资源、提示接入任意客户端。
  • Server 负责能力与权限，Client 负责连接，Host 决定把什么放进模型上下文。
  • 工具描述、返回大小、错误结构，直接决定模型能不能用对。
  • 安全靠最小权限、认证、审计，协议本身不提供。

下一篇把多个 Agent 组织起来：主管、专家、交接契约与成本控制。

读完了？进度会保存在本地浏览器。

相关阅读

06.01从零实现 Agent 内核

不依赖框架，用约 300 行 Python 实现工具注册、ReAct 循环、流式事件、预算、重试、trace 与单元测试。

阅读 06.02从零搭建可评估 RAG

PDF 解析、语义切分、混合检索、重排、带引用生成，以及 50 条评估集和指标报告。

阅读 06.04多智能体主管系统

四个角色、结构化交接契约、共享状态、并行执行、成本对比与错误传播演练。

阅读
内容会持续修订。发现错误或想补充，欢迎反馈。 遇到问题？查排查手册 →