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

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

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

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

# 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, '张三', '[email protected]', '上海'),
    (2, '李四', '[email protected]', '北京'),
    (3, '王五', '[email protected]', '深圳');

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 初始化完成")

python seed.py

# 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)

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

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 "未找到"

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}"

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

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} 行。"

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

npx @modelcontextprotocol/inspector python server.py

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

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

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

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