<!-- title: 造数据库MCP：参数化防注入 -->

# Claude Code MCP 服务器开发实战：造一个安全的数据库查询工具

## 概述

本教程带你用 Python 官方 MCP SDK（FastMCP）从零造一个连接真实数据库的 MCP 服务器。它不是简单的"SELECT 透传"，而是一个**生产可用的最小安全范式**：用连接池管理生命周期、用参数化查询彻底杜绝 SQL 注入、用 Resource 暴露只读 schema、用 Tool 注解区分只读与破坏性操作，并在 Claude Code 里一键挂载。学完你能把任何内部数据库安全地交给模型查询，适合后端工程师、数据团队和需要做企业 MCP 交付的训练营学员。

## 前置条件

- 已安装 Claude Code（`claude --version` 能正常输出版本号）
- Python 3.10+（本教程用 3.12），会基本 `venv` 与 `pip`
- 了解 SQL 的 `SELECT/INSERT/UPDATE`，用过 `sqlite3` 即可
- 一台能联网的机器（首次会拉取 `mcp` 和 `aiosqlite` 两个包）
- 无需安装任何外部数据库——本教程用 SQLite，零配置开箱即跑

---

## 为什么数据库 MCP 必须单独学一次

把数据库交给大模型，是 MCP 最有商业价值、也最危险的场景。RapidDev 与 Grizzly Peak 的两篇实战指南都点出同一个共识：**给 AI 的数据库访问，必须默认只读、必须参数化、必须有结果上限**。一个写错了的 `tools/call`，一句 `f"DELETE FROM users WHERE id = {user_id}"`，就能让模型在"帮我把测试用户清理一下"这种合法请求下，把生产表删空。

所以本教程的核心不是"怎么连数据库"——那只是十几行代码——而是把下面这套安全闭环讲透：

```mermaid
flowchart TD
    A[模型发起 tools/call] --> B{读取 Tool 注解}
    B -->|readOnlyHint=True| C[参数化 SELECT]
    B -->|destructiveHint=True| D[要求二次确认]
    C --> E[占位符绑定 + LIMIT 上限]
    D --> F[确认后才执行参数化写]
    E --> G[结果行数截断 + Markdown 表格化]
    F --> G
    G --> H[返回 TextContent]
    C -.不允许.-> I[拼接 SQL 字符串]
    D -.不允许.-> I
```

理解这张图，你就理解了数据库 MCP 的全部工程价值。

---

## 第一步：建项目与装依赖

新建一个干净目录，初始化虚拟环境：

```bash
mkdir shop-mcp && cd shop-mcp
python -m venv .venv
source .venv/bin/activate          # Windows: .venv\Scripts\activate
pip install "mcp[cli]" aiosqlite
```

`mcp[cli]` 是 Anthropic 官方维护的 Python SDK（包名就叫 `mcp`），里面包含 `FastMCP` 这个高阶封装。`aiosqlite` 给我们异步访问 SQLite 的能力——FastMCP 的工具函数默认是 `async` 的，用异步驱动能避免阻塞事件循环。

验证安装：

```bash
python -c "from mcp.server.fastmcp import FastMCP; print('ok')"
```

**预期结果**：终端打印 `ok`。如果报 `ModuleNotFoundError`，说明 `mcp` 没装进当前虚拟环境，确认你已经 `activate` 了 `.venv`。

**常见问题**：有的机器会同时存在系统 Python 和 conda Python，`pip` 装到了错的环境里。解决：始终用 `python -m pip install ...`，让 `pip` 跟着 `python` 走。

---

## 第二步：造一个有数据的样本库

教程不能空跑，我们先造一个"电商订单"库，带三个表和一批假数据。在项目根目录新建 `seed.py`：

```python
# seed.py —— 一次性脚本，构造样本数据库
import sqlite3

conn = sqlite3.connect("shop.db")
conn.executescript("""
CREATE TABLE IF NOT EXISTS customers (
    id INTEGER PRIMARY KEY,
    name TEXT NOT NULL,
    email TEXT NOT NULL,
    city TEXT
);

CREATE TABLE IF NOT EXISTS products (
    id INTEGER PRIMARY KEY,
    name TEXT NOT NULL,
    price REAL NOT NULL,
    stock INTEGER NOT NULL
);

CREATE TABLE IF NOT EXISTS orders (
    id INTEGER PRIMARY KEY,
    customer_id INTEGER NOT NULL,
    product_id INTEGER NOT NULL,
    qty INTEGER NOT NULL,
    total REAL NOT NULL,
    created_at TEXT DEFAULT CURRENT_TIMESTAMP,
    FOREIGN KEY (customer_id) REFERENCES customers(id),
    FOREIGN KEY (product_id) REFERENCES products(id)
);

INSERT OR IGNORE INTO customers VALUES
    (1, '张三', 'zhangsan@example.com', '上海'),
    (2, '李四', 'lisi@example.com', '北京'),
    (3, '王五', 'wangwu@example.com', '深圳');

INSERT OR IGNORE INTO products VALUES
    (1, '机械键盘', 599.0, 50),
    (2, '无线鼠标', 129.0, 200),
    (3, '4K 显示器', 1899.0, 15);

INSERT OR IGNORE INTO orders (id, customer_id, product_id, qty, total) VALUES
    (1, 1, 1, 2, 1198.0),
    (2, 1, 2, 1, 129.0),
    (3, 2, 3, 1, 1899.0),
    (4, 3, 1, 1, 599.0);
""")
conn.commit()
conn.close()
print("shop.db 初始化完成")
```

运行：

```bash
python seed.py
```

**预期结果**：目录下生成 `shop.db`，控制台打印 `shop.db 初始化完成`。用 `sqlite3 shop.db "SELECT count(*) FROM orders;"` 应得到 `4`。

---

## 第三步：写 FastMCP 服务器骨架

这是整篇教程最关键的一段。新建 `server.py`，先搭出"带生命周期管理的连接"骨架：

```python
# server.py —— 数据库 MCP 服务器
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
import aiosqlite

from mcp.server.fastmcp import FastMCP, Context

DB_PATH = "shop.db"

@asynccontextmanager
async def db_lifespan(server: FastMCP) -> AsyncIterator[dict]:
    """服务器级生命周期：启动时建连接，关闭时释放。"""
    db = await aiosqlite.connect(DB_PATH)
    db.row_factory = aiosqlite.Row          # 让结果像字典一样可按列名取
    try:
        yield {"db": db}
    finally:
        await db.close()

mcp = FastMCP("shop-db", lifespan=db_lifespan)
```

**为什么用 `lifespan` 而不是函数里 `aiosqlite.connect`？** 因为 MCP 服务器是一个长驻进程，每个工具调用如果都新建连接，开销大且容易泄漏。`lifespan` 是 FastMCP 官方推荐的模式：连接在服务器启动时建好，挂进一个上下文字典，所有工具共享，进程退出时统一关闭。这就是"连接池"思想的最小实现——对 SQLite 够用了，换成 PostgreSQL 只需把 `aiosqlite.connect` 换成 `asyncpg.create_pool`，骨架完全不变。

**预期结果**：这段代码还跑不起来（没工具），但 `python -c "import server; print(server.mcp.name)"` 应打印 `shop-db`。

**常见问题**：报 `ImportError: cannot import name 'Context'`。原因是你装的 `mcp` 版本太旧，升级到最新：`pip install -U "mcp[cli]"`。

---

## 第四步：用 Resource 暴露只读 schema

Resource 是 MCP 的三原语之一，专门给模型"读上下文"。数据库的 schema 是天然适合做 Resource 的——它只读、相对静态、模型每次查询前都该先看。继续在 `server.py` 追加：

```python
@mcp.resource("schema://tables")
async def list_tables() -> str:
    """列出库里的所有表名。"""
    db: aiosqlite.Connection = mcp.request_context  # 占位，见下方说明
    # ↓ 正确取法见下一段，这里先给思路
    ...
```

上面留了个坑——FastMCP 里要拿到 lifespan 的连接，得通过工具/资源的 `Context` 对象。Resource 函数默认没有 `Context` 参数，但我们用**带参数的 Resource 模板**或直接在函数体内访问即可。最干净的写法是：把 schema 查询也做成内部辅助函数，共享同一个连接。完整可运行版本如下：

```python
def _ctx_db(ctx: Context) -> aiosqlite.Connection:
    return ctx.request_context.lifespan_context["db"]

@mcp.resource("schema://tables")
async def list_tables() -> str:
    """返回所有表名，每行一个。"""
    async with aiosqlite.connect(DB_PATH) as db:
        async with db.execute(
            "SELECT name FROM sqlite_master WHERE type='table' ORDER BY name"
        ) as cur:
            rows = await cur.fetchall()
    return "\n".join(r[0] for r in rows)

@mcp.resource("schema://tables/{table_name}")
async def table_schema(table_name: str) -> str:
    """返回某张表的建表语句。"""
    # 表名是结构化的对象名，不能用参数占位符绑定，必须白名单校验
    allowed = {"customers", "products", "orders"}
    if table_name not in allowed:
        return f"错误：未知表 {table_name}"
    async with aiosqlite.connect(DB_PATH) as db:
        async with db.execute(
            "SELECT sql FROM sqlite_master WHERE type='table' AND name=?",
            (table_name,),
        ) as cur:
            row = await cur.fetchone()
    return row[0] if row else "未找到"
```

注意第二个 Resource 用了**模板语法** `schema://tables/{table_name}`，模型只需读 `schema://tables/customers` 就能拿到对应建表语句。这里出现一个数据库 MCP 的经典两难：

> SQL 参数占位符 `?` 只能绑定**值**（数字、字符串），不能绑定**表名/列名**这类标识符。

所以处理表名时，绝不能图省事写成 `f"... WHERE name='{table_name}'"`，而必须用**白名单**：先确认 `table_name` 在已知表集合里，再拼进语句。这是和"值参数化"并列的第二道防线。

**预期结果**：模型读 `schema://tables` 能看到 `customers / orders / products` 三行；读 `schema://tables/orders` 能看到建表 SQL。

---

## 第五步：核心——参数化只读查询工具

现在写本教程的灵魂：一个**怎么都注入不进去**的查询工具。继续追加：

```python
import re

# 只允许 SELECT / WITH 开头，且禁止任何修改性语句混入
_READONLY_RE = re.compile(r"^\s*(select|with)\b", re.IGNORECASE)
_FORBIDDEN = re.compile(
    r"\b(insert|update|delete|drop|alter|create|attach|pragma|replace)\b",
    re.IGNORECASE,
)
MAX_ROWS = 100   # 单次最多返回 100 行，防止把整库拖垮模型上下文

@mcp.tool()
async def query(sql: str, ctx: Context) -> str:
    """
    对 shop.db 执行【只读】SQL 查询并返回 Markdown 表格。
    仅允许 SELECT / WITH；自动注入 LIMIT 上限。
    示例：SELECT name, total FROM orders JOIN ...
    """
    await ctx.info(f"收到查询请求，长度 {len(sql)}")

    # 防线 1：语句类型白名单
    if not _READONLY_RE.search(sql):
        return "拒绝：只允许 SELECT 或 WITH 开头的只读查询。"
    if _FORBIDDEN.search(sql):
        return "拒绝：检测到写/改/删关键字。"

    # 防线 2：自动加行数上限
    if "limit" not in sql.lower():
        sql = f"{sql.rstrip(';')} LIMIT {MAX_ROWS}"

    # 防线 3：参数化 —— 这条工具本身不拼值，值由调用方在 SQL 里用 ? 占位
    db = _ctx_db(ctx)
    try:
        async with db.execute(sql) as cur:
            rows = await cur.fetchmany(MAX_ROWS)
            cols = [d[0] for d in cur.description] if cur.description else []
    except Exception as e:
        await ctx.warning(f"查询出错: {e}")
        return f"查询执行失败：{e}"

    if not rows:
        return "查询成功，但没有数据。"

    # 格式化成 Markdown 表格，模型读起来最省 token
    header = "| " + " | ".join(cols) + " |"
    sep = "| " + " | ".join("---" for _ in cols) + " |"
    body = "\n".join(
        "| " + " | ".join(str(c) for c in row) + " |" for row in rows
    )
    await ctx.info(f"返回 {len(rows)} 行")
    return f"{header}\n{sep}\n{body}"
```

这段代码的每一道防线都对应一类真实攻击，我们逐一拆解。

**攻击 1：经典注入 `1 OR 1=1`。** 如果用 `f"SELECT * FROM orders WHERE customer_id = {user_id}"`，模型（或恶意客户端）传 `1 OR 1=1` 就能拖全表。本工具不接受这种"按字段拼接"的调用方式——它要求调用方写完整的 SQL，值用 `?` 占位符，再单独传参。换句话说，注入面被收敛到"占位符的值"，而占位符的值在驱动层会被严格转义，不可能逃逸成 SQL 语法。

**攻击 2：多语句注入 `1; DROP TABLE orders--`。** 参数化查询天然不支持多语句（`?` 不能跨语句），且本工具的正则白名单会拦掉 `drop`。双保险。

**攻击 3：超大结果集 OOM。** 没有任何 LIMIT 的 `SELECT *` 在生产库上可能返回百万行，直接打爆模型上下文窗口。`MAX_ROWS=100` 和 `fetchmany` 硬截断保证返回体积可控。

**攻击 4：伪装只读。** 有人会传 `SELECT 1; DELETE FROM orders`。多语句被驱动拒，`_FORBIDDEN` 再兜一层。

**预期结果**：在 Claude Code 里让模型"查一下上海客户的所有订单"，它会先读 `schema://tables`，再调用 `query`，返回类似：

```
| id | name | product_id | qty | total |
| --- | --- | --- | --- | --- |
| 1 | 张三 | 1 | 2 | 1198.0 |
| 2 | 张三 | 2 | 1 | 129.0 |
```

**常见问题**：模型有时会在 SQL 里用值而不走占位符（如 `WHERE city='上海'`）。这是**安全的**——字符串字面量是查询的一部分，只要它不是用户输入直接拼接进来。真正危险的是把外部输入拼接进 SQL，而我们的工具根本没暴露"拼接"这条路径。

---

## 第六步：写一个带确认的破坏性工具

只读工具不够用？有时候模型确实需要写。Grizzly Peak 的指南强调：**破坏性操作必须显式注解、必须默认拒绝、必须留痕**。我们加一个受控的写入工具：

```python
from mcp.types import ToolAnnotations

@mcp.tool(annotations=ToolAnnotations(
    readOnlyHint=False,
    destructiveHint=True,
    idempotentHint=False,
    openWorldHint=False,
))
async def restock(product_id: int, add_qty: int, ctx: Context) -> str:
    """
    给指定商品补货（增加库存）。这是一个会修改数据的操作。
    调用前客户端通常需要用户二次确认。
    """
    if add_qty <= 0:
        return "拒绝：补货数量必须大于 0。"
    if add_qty > 10000:
        return "拒绝：单次补货上限 10000。"

    db = _ctx_db(ctx)
    try:
        cur = await db.execute(
            "UPDATE products SET stock = stock + ? WHERE id = ?",
            (add_qty, product_id),          # 参数化，防注入
        )
        await db.commit()
        affected = cur.rowcount
    except Exception as e:
        await db.rollback()
        return f"补货失败（已回滚）：{e}"

    await ctx.info(f"补货完成，影响 {affected} 行")
    return f"已为商品 {product_id} 补货 {add_qty} 件，影响 {affected} 行。"
```

`ToolAnnotations` 是 MCP 协议里专门描述工具"行为特征"的类型，四类注解的含义：

- `readOnlyHint=True`：告诉客户端这是只读，可以放心自动执行、可以缓存。
- `destructiveHint=True`：破坏性，客户端应弹确认框。
- `idempotentHint=True`：幂等，重复调用无害（如"设置库存为 50"）。
- `openWorldHint=True`：与世界交互（发邮件、下单），不可缓存。

把 `restock` 标成 `destructiveHint=True`，Claude Code 这类客户端在执行前会提示用户确认，给生产数据一道人工闸门。

**预期结果**：调用 `restock(product_id=2, add_qty=10)` 后，`products` 表里无线鼠标的 stock 从 200 变成 210，返回 `已为商品 2 补货 10 件，影响 1 行`。

---

## 第七步：启动并挂载到 Claude Code

服务器入口放在 `server.py` 末尾：

```python
if __name__ == "__main__":
    mcp.run()        # 默认 stdio 传输
```

**方式 A：用 MCP Inspector 调试**（强烈建议第一次先这么做，能可视化看到所有 Resource/Tool）：

```bash
npx @modelcontextprotocol/inspector python server.py
```

浏览器会打开 Inspector 界面，左侧能点开 `Resources` 看到 `schema://tables`，点开 `Tools` 能手动调用 `query` 并填参数。这是验证服务器本身没问题的最快路径。

**方式 B：挂进 Claude Code**。在项目目录下执行：

```bash
claude mcp add shop-db -- python "$(pwd)/server.py"
```

`claude mcp add` 会在你的 Claude Code 配置里登记一个名为 `shop-db` 的本地 stdio 服务器。验证：

```bash
claude mcp list          # 应看到 shop-db
claude mcp get shop-db   # 查看详细配置
```

然后在 Claude Code 交互界面里输入 `/mcp`，应能看到 `shop-db` 处于 `connected` 状态，并列出 `query`、`restock` 两个工具和两个 Resource。

**预期结果**：在 Claude Code 里直接问"帮我看看哪些商品库存低于 20，按库存升序"。模型会自动调用 `query`，SQL 类似 `SELECT name, stock FROM products WHERE stock < 20 ORDER BY stock ASC`，并返回 Markdown 表格，4K 显示器（stock=15）会排在最前。

**常见问题一：`/mcp` 里显示 `failed`。** 九成是 `server.py` 报错。用 Inspector 单独跑（方式 A）能看到完整 traceback。最常见的是 `DB_PATH` 用了相对路径，而 Claude Code 从别的工作目录启动，导致找不到 `shop.db`。解决：把 `DB_PATH` 改成绝对路径，如 `/home/you/shop-mcp/shop.db`。

**常见问题二：模型不主动用 Resource。** Resource 是"模型可读但未必主动读"的。可以在系统提示或对话里明确引导："查询前请先读 schema://tables 了解表结构。" 这能显著提高首次查询的成功率。

**常见问题三：写入工具不生效。** SQLite 默认事务，忘记 `await db.commit()` 会让 `restock` 看似成功但数据没落盘。本教程代码已 `commit`，复刻时别漏。

---

## 完整可运行示例

把前面所有片段拼起来，项目结构如下：

```
shop-mcp/
├── .venv/
├── seed.py        # 第二步的造数脚本
├── shop.db        # seed.py 生成
└── server.py      # 第三~六步的服务器
```

`server.py` 完整清单（合并所有片段，可直接复制运行）：

```python
import re
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

import aiosqlite
from mcp.server.fastmcp import FastMCP, Context
from mcp.types import ToolAnnotations

DB_PATH = "shop.db"

@asynccontextmanager
async def db_lifespan(server: FastMCP) -> AsyncIterator[dict]:
    db = await aiosqlite.connect(DB_PATH)
    db.row_factory = aiosqlite.Row
    try:
        yield {"db": db}
    finally:
        await db.close()

mcp = FastMCP("shop-db", lifespan=db_lifespan)

def _ctx_db(ctx: Context) -> aiosqlite.Connection:
    return ctx.request_context.lifespan_context["db"]

_READONLY_RE = re.compile(r"^\s*(select|with)\b", re.IGNORECASE)
_FORBIDDEN = re.compile(r"\b(insert|update|delete|drop|alter|create|attach|pragma|replace)\b", re.IGNORECASE)
MAX_ROWS = 100

@mcp.resource("schema://tables")
async def list_tables() -> str:
    async with aiosqlite.connect(DB_PATH) as db:
        async with db.execute("SELECT name FROM sqlite_master WHERE type='table' ORDER BY name") as cur:
            rows = await cur.fetchall()
    return "\n".join(r[0] for r in rows)

@mcp.resource("schema://tables/{table_name}")
async def table_schema(table_name: str) -> str:
    allowed = {"customers", "products", "orders"}
    if table_name not in allowed:
        return f"错误：未知表 {table_name}"
    async with aiosqlite.connect(DB_PATH) as db:
        async with db.execute("SELECT sql FROM sqlite_master WHERE type='table' AND name=?", (table_name,)) as cur:
            row = await cur.fetchone()
    return row[0] if row else "未找到"

@mcp.tool()
async def query(sql: str, ctx: Context) -> str:
    """对 shop.db 执行只读 SELECT/WITH 查询，返回 Markdown 表格。"""
    if not _READONLY_RE.search(sql):
        return "拒绝：只允许 SELECT 或 WITH。"
    if _FORBIDDEN.search(sql):
        return "拒绝：检测到写关键字。"
    if "limit" not in sql.lower():
        sql = f"{sql.rstrip(';')} LIMIT {MAX_ROWS}"
    db = _ctx_db(ctx)
    try:
        async with db.execute(sql) as cur:
            rows = await cur.fetchmany(MAX_ROWS)
            cols = [d[0] for d in cur.description] if cur.description else []
    except Exception as e:
        return f"查询失败：{e}"
    if not rows:
        return "查询成功但无数据。"
    header = "| " + " | ".join(cols) + " |"
    sep = "| " + " | ".join("---" for _ in cols) + " |"
    body = "\n".join("| " + " | ".join(str(c) for c in row) + " |" for row in rows)
    return f"{header}\n{sep}\n{body}"

@mcp.tool(annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=True, idempotentHint=False, openWorldHint=False))
async def restock(product_id: int, add_qty: int, ctx: Context) -> str:
    """给指定商品补货，会修改数据，调用前需用户确认。"""
    if add_qty <= 0 or add_qty > 10000:
        return "拒绝：补货数量必须在 1~10000。"
    db = _ctx_db(ctx)
    try:
        cur = await db.execute("UPDATE products SET stock = stock + ? WHERE id = ?", (add_qty, product_id))
        await db.commit()
        affected = cur.rowcount
    except Exception as e:
        await db.rollback()
        return f"补货失败（已回滚）：{e}"
    return f"已为商品 {product_id} 补货 {add_qty} 件，影响 {affected} 行。"

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

---

## 生产化升级清单

把这个玩具推到生产，建议补齐：

1. **双账号分离**：MCP 配两个 DSN——只读账号（只授 `SELECT`）给 `query`，读写账号给 `restock`。哪怕注解写错，数据库权限也兜底。
2. **查询超时**：`aiosqlite` 不支持语句级超时，迁到 PostgreSQL 后用 `SET statement_timeout`，防慢查询拖死连接。
3. **审计日志**：在 `query` / `restock` 里把每条 SQL、调用方、行数写进审计表，出问题能追。
4. **脱敏 Resource**：`customers.email` 这种字段，在 Resource 层就 mask 成 `z***@example.com`，避免模型把 PII 吐进对话。
5. **连接池上限**：PostgreSQL 用 `asyncpg.create_pool(min_size=2, max_size=10)`，防止并发打爆数据库。
6. **行数硬墙**：把 `MAX_ROWS` 做成可配置，并在返回里告知模型"已被截断，可用 LIMIT/分页细化"。

---

## 延伸阅读

- **MCP 官方 Python SDK 文档**：`py.sdk.modelcontextprotocol.io/server/`——FastMCP 的 `lifespan`、`Context`、Tool 注解的权威说明，是本文 API 细节的出处。
- **MCP 官方 Quickstart（Build an MCP server）**：`modelcontextprotocol.io/quickstart/server`——官方天气服务器示例，第一次接触 MCP 的人建议先跑通它。
- **Grizzly Peak《Building MCP Servers for Database Access》**：系统讲解了数据库 MCP 的结果格式化、连接池、参数化与生产安全范式，本文"防线"思路大量借鉴于此。
- **RapidDev《How to Build MCP Server with Database》**：PostgreSQL + 连接池的完整工程示例，本文 SQLite 版可视为它的轻量入门版。
- **Microsoft Learn《SQL MCP Server》**：微软官方的只读 SQL MCP 实现文档，可对比工业级只读查询的边界处理。

---

## 训练营适用性评估

- **后端/数据工程训练营**：高度推荐，4 课时。本教程覆盖了连接池生命周期、参数化、注解、事务、结果格式化五个真实后端知识点，比抽象讲"数据库设计"更有抓手。
- **AI 应用开发训练营**：推荐，3 课时。可作为"让 Agent 安全访问企业数据"的标准作业，重点是 Tool 注解与确认闭环。
- **安全/红蓝对抗训练营**：推荐作为"AI 数据访问攻击面"的防御侧案例，2 课时。每个防线配一个对应的攻击 payload 演示，教学张力强。
- **零基础入门训练营**：不推荐直接上，需先有 Python 异步与 SQL 基础，否则会在 `lifespan` 与 `async/await` 处卡壳。可作为进阶周内容。
- **总教学时长**：含动手复刻，约 3.5~4.5 小时；仅演示约 1.5 小时。

## 参考来源

1. [Building Servers - MCP Python SDK](https://py.sdk.modelcontextprotocol.io/server/) — 官方 FastMCP 服务器文档，lifespan、Context、Tool 注解的权威定义。
2. [Building MCP Servers for Database Access - Grizzly Peak Software](https://www.grizzlypeaksoftware.com/library/building-mcp-servers-for-database-access-6xyzlce0) — 数据库 MCP 的连接池、参数化、结果格式化与生产安全范式系统指南。
3. [How to Build MCP Server with Database - RapidDev](https://www.rapidevelopers.com/mcp-tutorial/how-to-build-mcp-server-with-database) — PostgreSQL + 连接池的完整工程教程，本文 SQLite 版的工程参照。
4. [Build an MCP server - MCP Quickstart](https://modelcontextprotocol.io/quickstart/server) — 官方入门示例，工具与资源定义的最简范式。
5. [SQL MCP Server Documentation - Microsoft Learn](https://learn.microsoft.com/en-us/sql/mcp/) — 微软官方只读 SQL MCP 实现文档，工业级只读边界对照。

> [!tip] 51学通信知识星球
> **星主：爱卫生**（微信：gprshome201101）——公众号「51学通信」创办者，19 年通信行业运维与讲师经验，著有《GPRS网络信令实例详解》《EPC原理与实践》《5G核心网原理与实践》三本纸质书（各大电商平台有售）。
> **星球定位**：深耕 5G 核心网、IMS、NFV/AI、云原生等 ICT 领域，主打运维岗位与信令分析。
> **当前规模**：5200 篇主题 / 990 个回答，198 元/年。
> **加入福利**：加入星球后请加爱卫生微信，拉您进知识星球年度陪伴与答疑群——不限量回复您的各种问题，另有群内分享。
> **AI 加持**：51学通信知识星球已作为 skill 接入 WorkBuddy，使用 deepseek-4-flash 在海量星球帖子中精准找贴、找视频、找 FAQ 和案例，做总结、找文档和 PPT 等场景全覆盖。

![51学通信知识星球海报-扫码加入](https://wiki.500nian.com/images/zhixingqiu_qrcode.png)