MCP 数据库集成

Model Context Protocol (MCP) 入门

James Chapman

AI Curriculum Manager, DataCamp

为何为时区服务器添加数据库？

查询能力：无需加载全量数据即可筛选、搜索和排序
可扩展性：借助数据库索引进行快速查找

数据库连接

为何为时区服务器添加数据库？

查询能力：无需加载全量数据即可筛选、搜索和排序
可扩展性：用数据库索引快速查找
并发：多次工具调用或客户端可安全访问数据库
生产就绪：备份、复制与单一事实来源

数据库优势

数据访问：工具 vs. 资源

工具图标

动态、强依赖用户输入、包含写入操作
示例：数据分析与运营

资源图标

只读、静态数据、参考信息
示例：指南、文档、策略

连接生命周期

数据库连接

数据库资源：timezone_server.py

from mcp.server.fastmcp import FastMCP
import sqlite3

mcp = FastMCP("Timezone Converter")

# Create connection at startup; reuse in handlers
conn = sqlite3.connect("timezones.db")
conn.row_factory = sqlite3.Row

@mcp.resource("db://timezones")
def get_locations() -> str:
    try:
        cursor = conn.execute("SELECT timezone FROM locations")
        rows = cursor.fetchall()
        return "\n".join(r["timezone"] for r in rows)
    except sqlite3.Error as e: return f"Error: {e}"

数据库查询工具：timezone_server.py

@mcp.tool()
def lookup_locations(prefix: str) -> str:
    """Find timezones that contain the given prefix."""
    try:
        cursor = conn.execute(
            "SELECT timezone FROM locations WHERE timezone LIKE ? LIMIT 50", (f"%{prefix}%",))
        rows = cursor.fetchall()
        return "\n".join(r["timezone"] for r in rows)
    except sqlite3.Error as e: return f"Error: {e}"


if __name__ == "__main__":
    try:
        mcp.run(transport="stdio")
    finally:
        conn.close()

安全与生产实践

参数化（?）查询：切勿将用户或 LLM 输入直接拼接进 SQL
- 提示注入 → 通过提示恶意注入代码

SELECT timezone FROM locations WHERE timezone LIKE ? LIMIT 50

对 MCP 服务器优先使用只读、最小权限的数据访问
设置限制：最大行数与超时

Passons à la pratique !

Model Context Protocol (MCP) 入门