生命不息，折腾不止。上一篇的 MCP Server 只会算加法，这一篇把它变成真能干活的：接上真实的 SQLite 和文件目录，把 Resources 的 URI 模板讲透，再把 Server 从本地 stdio 换成 Streamable HTTP 加鉴权挂到公网，让 dsh、Claude Code 几个客户端共用同一套工具。

上一篇（MCP 实战第一课）我们把 MCP 是什么、三个原语、客户端-服务器架构讲清了，也跑通了一个只会 add(a, b) 的玩具 Server。说实话，玩具跑通那一下爽，但它连你硬盘上的一个文件都读不到，更别说数据库。

这一篇就解决「玩具到真家伙」的距离。全程还是 Python，还是当前稳定版的 mcp（2.3.0，协议修订 2026-07-28），命令我都实机跑过或者对着官方文档核过，拿不准的地方我会标出来。

一、先想明白：从玩具到真家伙，差三块

别看代码量不大，一个能用的 Server 和玩具之间隔着三件事，缺一件都上不了生产：

  1. 数据层——它得能碰到你真实的数据。SQLite、MySQL、一堆 Markdown 文件、公司内部 API，随便哪个都行，但不能再是内存里写死的变量。
  2. 协议层——数据一多，resources/list 一次性返回几万条会把上下文撑爆。这时候要懂 Resources 的 URI 模板（一个函数服务一万个 URI）和分页（cursor 机制）。
  3. 部署层——想给多个客户端共用，就得把 stdio（本地子进程）换成 Streamable HTTP（一个端口），并且必须有门锁。一个裸奔在公网上的 MCP Server，等于把数据库的读写权限挂到互联网上任人调用。

记住这三个词：数据、协议、门锁。下面三节正好一个对一个。

顺便提醒一句承接上一篇的坑：v2 里入口是 from mcp.server import MCPServer，网上大量旧教程还在写 from mcp.server.fastmcp import FastMCP，照抄直接 ModuleNotFoundError。

二、接真实数据：三十行把 SQLite 变成 Resources

先来个最朴素的真实场景：一个笔记库。装好环境（Python 3.10+）：

1
2
pip install "mcp[cli]"
# 或者 uv add "mcp[cli]"

准备一份数据（放在跟 server.py 同一个目录）：

1
2
3
4
5
6
7
8
9
10
11
12
13
# init_db.py —— 只跑一次
import sqlite3

conn = sqlite3.connect("notes.db")
conn.execute("CREATE TABLE IF NOT EXISTS notes (id INTEGER PRIMARY KEY, title TEXT, body TEXT)")
conn.executemany(
    "INSERT INTO notes (title, body) VALUES (?, ?)",
    [("MCP 是什么", "一个让模型安全拿到上下文的协议。"),
     ("Streamable HTTP", "MCP 现在推荐的远程传输方式，取代了老的 SSE。"),
     ("鉴权怎么做", "服务器当资源服务器，只验 token，不发 token。")],
)
conn.commit()
conn.close()

然后是主角 server.py：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
import sqlite3
from mcp.server import MCPServer

DB = "notes.db"
mcp = MCPServer("Notes")


def q(sql: str, args: tuple = ()) -> list[dict]:
    with sqlite3.connect(DB) as conn:
        conn.row_factory = sqlite3.Row
        return [dict(r) for r in conn.execute(sql, args)]


@mcp.resource("db://notes")
def list_notes() -> list[dict]:
    """全部笔记的概览（id + 标题）。"""
    return q("SELECT id, title FROM notes ORDER BY id")


@mcp.resource("db://notes/{note_id}")
def get_note(note_id: int) -> dict:
    """按 id 读一篇笔记的完整内容。"""
    rows = q("SELECT * FROM notes WHERE id = ?", (note_id,))
    return rows[0] if rows else {"error": f"note {note_id} 不存在"}


@mcp.tool()
def search_notes(keyword: str, limit: int = 10) -> list[dict]:
    """按关键词搜索笔记标题与正文，返回匹配条目。"""
    like = f"%{keyword}%"
    return q(
        "SELECT id, title FROM notes WHERE title LIKE ? OR body LIKE ? LIMIT ?",
        (like, like, limit),
    )


if __name__ == "__main__":
    mcp.run()  # 默认 stdio，先本地调试

跑一下看效果（mcp dev 会拉起 MCP Inspector 网页界面，能点着看 Resources 和 Tools）：

1
uv run mcp dev server.py

几个必须记住的点：

  • Resources 是给「应用程序」读的，Tools 是给「模型」调的。 笔记列表、单篇内容这种只读的，放 Resources；搜索、写笔记这种带动作的，放 Tools。别把写操作塞进 resource，它本来就不是干这个的。
  • 返回值自动转型：返回 str 是文本，返回 dict/list 自动序列化成 JSON 文本，返回 bytes 变成 base64 blob。要标类型就用 @mcp.resource("docs://readme", mime_type="text/markdown")。
  • UUID 里带 {placeholder} 就是模板，会被列进 resources/templates/list 而不是 resources/list。
  • 函数是在「读」的那一刻才执行，不是列清单的时候。所以读到不存在的 id，rows[0] 那种写法要自己兜住。

如果你想让这个 Server 顺带做个「AI 摘要」的工具（比如把一篇笔记丢给模型浓缩成一句话），这就是接模型的入口——写个 @mcp.tool() 在里面调 OpenAI 兼容接口即可，我自己的做法是指向 ai.aklibk.com，一个 key 就能在多模型之间切换；要接 Claude/GPT/Gemini 这类国外模型，走中转的好处是国内直连、免绑卡、人民币按量付费，省得为调一次接口折腾一堆东西。

三、URI 模板：一个函数服务一万个 URI

上面 db://notes/{note_id} 只是最基础的形式。MCP 的模板语法用的是 RFC 6570，SDK 实现了其中够用的一部分，一共就五个操作符：

模板写法 客户端读的 URI 你的函数收到
books://{isbn} books://978-0441172719 "978-0441172719"
books://{isbn} books://978/extra 不匹配（{name} 遇 / 就停）
manuals://{+path} manuals://printing/setup.md "printing/setup.md"（保留斜杠）
reviews://{isbn}{?limit,sort} reviews://978...?sort=top isbn、limit、sort
shelves://browse{/path*} shelves://browse/fiction/sci-fi ["fiction", "sci-fi"]

几个实战里最容易踩的点：

  1. {name} 只吃一段路径，{+name} 才保留斜杠。 你写文件路径、嵌套对象键，一律用 {+name}，否则 docs/intro.md 这种多级路径根本匹配不上。
  2. 类型标注会自动转换。 参数写 note_id: int，读 db://notes/42 进来就是整数 42，不用自己 int()。
  3. 查询参数必须给 Python 默认值。 {?limit,sort} 里的 limit 和 sort 得写成 limit: int = 10, sort: str = "newest"，因为它们匹配很宽松——客户端可以只传一个、顺序随意、多的直接忽略。少写默认值，SDK 在装饰器阶段就会直接 ValueError。
  4. 两个变量紧挨着不行。 manuals://{+path}{ext} 会被拒（没法判断边界），中间得有字面量分隔。

然后是安全。模板参数来自客户端，如果直接拼进文件路径，../../etc/passwd 就出去了。SDK 默认就拦住了这三类：含 .. 逃逸的、绝对路径（/etc/passwd、C:\Windows、C:foo 这种单字母加冒号的也拦）、带 null 字节的。而且它对解码后的值做检查，..%2Fetc、%2E%2E/etc 这些编码绕过一样拦。

但默认检查只是启发式前置过滤，真碰文件系统还得用 safe_join 兜底：

1
2
3
4
5
6
7
8
9
10
11
12
from pathlib import Path
from mcp.server import MCPServer
from mcp.shared.path_security import safe_join

DOCS = Path("./docs")
mcp = MCPServer("Docs")


@mcp.resource("docs://{+path}")
def read_doc(path: str) -> str:
    """读取 docs 目录下的文档，多级路径用 / 分隔。"""
    return safe_join(DOCS, path).read_text(encoding="utf-8")

safe_join 会把路径解析完再确认没跑出 DOCS 目录，符号链接逃逸、..、绝对路径都拦，逃出去就抛 PathEscapeError。反过来，如果某个参数确实要收绝对路径（比如导入工具），可以在装饰器上单独豁免：@mcp.resource("imports://preview/{+source}", security=ResourceSecurity(exempt_params={"source"}))。

四、列表太大怎么办：分页与 cursor

MCPServer 的默认行为很简单——每次 list_* 请求，它把全部东西一次性返回，next_cursor 永远是 None。几十个工具、几百条资源，这样最省事。

但你的笔记库有十万条呢？一次性序列化出来就是灾难。协议给的答案是 cursor（游标）：服务器返回一页加一个不透明 token，客户端把这个 token 原样送回来拿下一页。要分页就得写自己的 list handler，而 @mcp.resource() 装饰器没这个钩子，得用低级 Server：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from typing import Any
from mcp_types import ListResourcesResult, PaginatedRequestParams, Resource
from mcp.server import Server, ServerRequestContext

PAGE = 50


async def list_notes(ctx: ServerRequestContext[Any], params: PaginatedRequestParams | None) -> ListResourcesResult:
    start = 0 if params is None or params.cursor is None else int(params.cursor)
    rows = q("SELECT id, title FROM notes ORDER BY id LIMIT ? OFFSET ?", (PAGE, start))
    page = [Resource(uri=f"db://notes/{r['id']}", name=r["title"]) for r in rows]
    next_cursor = str(start + PAGE) if len(rows) == PAGE else None
    return ListResourcesResult(resources=page, next_cursor=next_cursor)


server = Server("Notes", on_list_resources=list_notes)

注意：低级 Server 上，handler 是构造函数的参数，不是装饰器。你上了这条路，on_list_tools、on_call_tool、on_read_resource 这些也得自己写——换来的是完全的控制权。

客户端那边 draining 一个分页列表就是一个 while True：

1
2
3
4
5
6
7
8
cursor = None
resources = []
while True:
    page = await client.list_resources(cursor=cursor)
    resources.extend(page.resources)      # 先收下这一页，再看有没有下一页
    if page.next_cursor is None:
        break
    cursor = page.next_cursor

三条规则记死：

  • cursor 是不透明的。 客户端只能原样回传上一页给的 next_cursor，绝不能自己拼一个。你 int("page-2") 试试，直接吃一个 MCPError(-32603)。
  • 页大小由服务端定。 协议里没有 limit= 参数，客户端说了不算。
  • 不分页的客户端照样能用。 它调一次拿到第一页，next_cursor 直接丢掉，只是看得少，不会崩。

五、挂到公网：Streamable HTTP + 鉴权

数据接好了，现在把它从「只能本地子进程跑」变成「一个 URL 谁都能连（但得有钥匙）」。

第一步，换传输。 就改 __main__ 那一行：

1
2
if __name__ == "__main__":
    mcp.run(transport="streamable-http", stateless_http=True, json_response=True)

服务起来后端点在 http://127.0.0.1:8000/mcp。stateless_http=True 加 json_response=True 是官方对生产部署的推荐配置，省掉会话状态、横向扩容友好。要嵌进已有的 Starlette/FastAPI 应用，用 mcp.streamable_http_app() 拿到一个标准 ASGI app 挂 Mount 就行；但记住两条坑：挂载后内置 lifespan 失效，宿主 app 的 lifespan 里必须自己 async with mcp.session_manager.run()，否则第一个请求就报 Task group is not initialized；另外它默认只认 localhost 的 Host 头（防 DNS rebinding），真机名部署要配：

1
2
3
4
5
6
7
from mcp.server.transport_security import TransportSecuritySettings

security = TransportSecuritySettings(
    allowed_hosts=["mcp.example.com", "mcp.example.com:*"],
    allowed_origins=["https://app.example.com"],
)
app = mcp.streamable_http_app(transport_security=security)

如果前面有 Nginx 反代、Host 头已经是你在控，干脆关掉更省事：TransportSecuritySettings(enable_dns_rebinding_protection=False)。不然客户端只会看到一句莫名其妙的 421 Misdirected Request，错误原因藏在服务端日志里。

第二步，上门锁。 在 OAuth 的语境里，你的 MCP Server 是资源服务器（Resource Server）：它从不登录用户、从不发 token，只干一件事——看每个请求 Authorization 头里的 token 对不对。SDK 把整个对接面收敛成一个 TokenVerifier：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
import os
from pydantic import AnyHttpUrl
from mcp.server import MCPServer
from mcp.server.auth.provider import AccessToken, TokenVerifier
from mcp.server.auth.settings import AuthSettings


class StaticTokenVerifier(TokenVerifier):
    async def verify_token(self, token: str) -> AccessToken | None:
        expected = os.environ["MCP_TOKEN"]          # 绝不硬编码
        if token and token == expected:
            return AccessToken(token=token, client_id="my-client", scopes=["notes:read"])
        return None


mcp = MCPServer(
    "Notes",
    token_verifier=StaticTokenVerifier(),
    auth=AuthSettings(
        issuer_url=AnyHttpUrl("https://auth.example.com"),
        resource_server_url=AnyHttpUrl("https://mcp.example.com/mcp"),
        required_scopes=["notes:read"],
    ),
)

三件事一起发生：

  • token_verifier 和 auth 必须成对出现，只给一个，MCPServer(...) 在还没开始服务之前就抛 ValueError。
  • 没带 token 或 token 不对，请求在门口就被拦下：401 加一个 WWW-Authenticate 头，头里 resource_metadata 指向 /.well-known/oauth-protected-resource/mcp，也就是 RFC 9728 的发现文档。客户端顺着这条线索自己找授权服务器、拿 token、重试——这段你一行代码没写，SDK 全给了。
  • handler 里想知道「谁在调」，直接 from mcp.server.auth.middleware.auth_context import get_access_token，拿到你自己构造的那个 AccessToken，client_id、scopes 都在，想按 scope 拒绝就按 scope 拒绝。

第三步，TLS 交给 Nginx。 峰哥式的标准做法，反代一段：

1
2
3
4
5
6
7
8
9
location /mcp {
    proxy_pass http://127.0.0.1:8000;
    proxy_http_version 1.1;
    proxy_set_header Host $host;
    proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
    proxy_set_header X-Forwarded-Proto $scheme;
    proxy_buffering off;          # SSE 流式响应不能缓冲
    proxy_read_timeout 300s;
}

两个必须交代的边界：静态 token 是给私有部署/个人工具用的，简单有效但不会过期、泄露了得手动换；面向公网、多用户的场景才需要上真 OAuth 2.1。另外 stdio 根本没有鉴权这回事——管道里没有 HTTP 头，token_verifier 永远不会被调用，本地 stdio 的安全边界就是「谁启动了那个进程」。

六、客户端怎么连：dsh 与 Claude Code

Server 上线了，最后一步是把它挂到你每天用的 Agent 上。

DeepSeek Harness（dsh）走官方 MCP 客户端插件，一个 server 一个插件实例，写在 cordis.patch.yml 里：

1
2
3
4
5
6
7
8
- id: mcp-notes
  name: '@deepseek-ai/dsh-mcp-client'
  config:
    serverName: notes
    transport: streamable-http
    url: https://mcp.example.com/mcp
    headers:
      Authorization: !!js '`Bearer ${process.env.MCP_TOKEN}`'

serverName 决定工具命名，注册进来就是 mcp__notes__search_notes 这种形式，所以改 serverName 等于改模型看到的每一个工具名，定好了别乱动。凭据用 Cordis 的 !!js 标签从环境变量取，别把真 token 写进要提交的 patch 文件。改这段配置会触发热重载——server 断开重连，dsh 进程本身不重启。

Claude Code 一条命令：

1
2
claude mcp add --transport http notes https://mcp.example.com/mcp \
  --header "Authorization: Bearer $MCP_TOKEN"

配完 claude mcp list 看状态：✔ Connected 才算真的通，! Needs authentication 说明它连上了但你没给对钥匙，✘ Failed to connect 才是根本没响应。

上线前先自测一次，两行命令就能判断门锁在不在位：

1
2
3
4
5
6
# 不带 token：期望 401（鉴权生效，不是 200 裸奔）
curl -s -o /dev/null -w "%{http_code}\n" https://mcp.example.com/mcp -X POST \
  -H 'Content-Type: application/json' -d '{"jsonrpc":"2.0","id":1,"method":"tools/list"}'

# 看发现文档：期望一份带 authorization_servers 的 JSON
curl -s https://mcp.example.com/.well-known/oauth-protected-resource/mcp

七、几个高频坑 + 一句话总结

  • ModuleNotFoundError: mcp.server.fastmcp：v2 已经改用 from mcp.server import MCPServer，钉住老版本就用 mcp>=1.28,<2 先止血。
  • token_verifier 配了但 auth 没配（或反过来）：构造函数直接 ValueError，不是运行时才炸。
  • 部署到真域名后全员 421：忘了 transport_security=，或反代改了 Host 却没允许。
  • Mount 进大应用后首个请求报 Task group is not initialized：宿主 lifespan 里没 mcp.session_manager.run()。
  • URI 带中文或空格：Streamable HTTP 会把 URI 镜像进每个 resources/read 的 Mcp-Name 头，非 ASCII 会被包成 =?base64?...?=，网关日志瞬间不可读。能 ASCII 就 ASCII。
  • 自定义路由 @mcp.custom_route() 永远不鉴权，健康检查放这儿可以，私密接口别放。

一句话收口：接真实数据（Resources + Tools 分清楚）、用好模板与分页（一个函数服务一万个 URI）、上线必上锁（Streamable HTTP + TokenVerifier + 反代 TLS），这三步走完，你手里就是一个能被多个 Agent 客户端共用的真实工具集了。

生命不息，折腾不止。下一篇「MCP 实战第三课」，我们转到客户端这一侧：把 Prompts 原语和 Completion 参数自动补全用起来，加上日志与进度上报（长任务不再干等），再在 dsh 和 Claude Code 里同时挂载多个 Server，拼出一套「本地文件 + 数据库 + 远程服务」混合工作流，配上排障清单——让 Agent 真的把这些工具用顺手，而不只是「连上了」。