FastMCP 实战：30 行 Python 给 AI 造一个数据库查询工具
上周团队来了个新需求：产品经理想在 Claude 里直接查询数据库中的用户反馈数据。起初的第一反应是——写个 API，再套一层 function calling 的封装。折腾了半天才发现，MCP + FastMCP 这条路其实要简单得多。仅需 30 多行代码，就能让 Claude Desktop 直接查询 SQLite 数据库了。
本文记录从零搭建的完整过程，包括实际踩过的坑和解决方案。
MCP 是 Anthropic 推出的一套开放协议，它的核心目标非常纯粹：为 AI 应用与外部工具之间提供一个统一的通信标准。你可以把它理解为 AI 世界的 USB-C 接口——无论是数据库、文件系统还是 API 服务，只要实现了 MCP 协议，Claude、VS Code、Cursor 等客户端都能直接连接使用。
该协议基于 JSON-RPC 2.0，分为两层：数据层负责消息格式与语义，传输层负责通信方式（本地 stdio 或远程 HTTP）。
FastMCP 是当下最主流的 Python MCP 框架。2024 年 FastMCP 1.0 被合并进官方 Python SDK，目前独立维护的版本日下载量已突破百万，约 70% 的 MCP Server 底层都在使用它。
环境准备
# 用 uv 安装（推荐）uv pip install fastmcp# 或者 pippip install fastmcp# 验证安装python -c "import fastmcp; print(fastmcp.__version__)"本地环境采用 Python 3.11 + macOS，FastMCP 3.x 版本。Windows 和 Linux 同样可以正常运行。
第一个 MCP Server：5 分钟快速上手
先来一个最简单的示例，构建一个能执行数学运算的工具：
# calc_server.pyfrom fastmcp import FastMCPmcp = FastMCP("计算器")@mcp.tooldef add(a: float, b: float) -> float:"""两个数相加"""return a + b@mcp.tooldef multiply(a: float, b: float) -> float:"""两个数相乘"""return a * bif __name__ == "__main__":mcp.run()启动运行：
python calc_server.py默认走 stdio 传输模式，此时 Server 等待客户端连接。如果想改用 HTTP：
python calc_server.py# 或者在代码里改成 mcp.run(transport="http", port=8000)FastMCP 自动完成了三件事：将函数名作为工具名称，将文档字符串作为工具描述，将参数类型标注转换为 JSON Schema。全程无需手动配置。
实战项目：为 AI 对接 SQLite 数据库
下面是我在实际项目中使用的代码，做了适当简化。需求是让 AI 能够查询用户反馈数据库。
# feedback_server.pyimport sqlite3from fastmcp import FastMCPDB_PATH = "feedback.db"mcp = FastMCP("用户反馈查询",instructions="这个服务器提供用户反馈数据的查询功能。" "用 search_feedback 按关键词搜索，用 get_stats 看整体统计。")def get_conn():conn = sqlite3.connect(DB_PATH)conn.row_factory = sqlite3.Rowreturn conn@mcp.tooldef search_feedback(keyword: str, limit: int = 20) -> list[dict]:"""按关键词搜索用户反馈。keyword 是搜索词，limit 控制返回数量，默认 20 条。"""conn = get_conn()rows = conn.execute("SELECT id, user_id, content, created_at, rating ""FROM feedback WHERE content LIKE ? ORDER BY created_at DESC LIMIT ?",(f"%{keyword}%", limit)).fetchall()conn.close()return [dict(r) for r in rows]@mcp.tooldef get_stats() -> dict:"""获取反馈数据的整体统计：总数、平均评分、最近 7 天数量。"""conn = get_conn()total = conn.execute("SELECT COUNT(*) FROM feedback").fetchone()[0]a vg_rating = conn.execute("SELECT A VG(rating) FROM feedback").fetchone()[0]recent = conn.execute("SELECT COUNT(*) FROM feedback ""WHERE created_at > datetime('now', '-7 days')").fetchone()[0]conn.close()return {"total": total,"a vg_rating": round(a vg_rating, 2) if a vg_rating else 0,"recent_7d": recent}@mcp.resource("data://schema")def get_schema() -> str:"""返回 feedback 表的建表语句，方便 AI 理解数据结构。"""conn = get_conn()schema = conn.execute("SELECT sql FROM sqlite_master WHERE type='table' AND name='feedback'").fetchone()conn.close()return schema[0] if schema else "表不存在"if __name__ == "__main__":mcp.run(transport="http", port=8000)这段代码涉及 MCP 的三个核心概念：
- Tool（工具）：
search_feedback和get_stats，AI 可主动调用以执行查询操作 - Resource（资源）：
get_schema，被动数据源，AI 可以读取但不会触发副作用 - instructions 参数：告知 AI 该 Server 的功能定位及使用方法
客户端连接：两种方式
方式一：配置到 Claude Desktop
编辑 ~/Library/Application Support/Claude/claude_desktop_config.json：
{"mcpServers": {"feedback": {"command": "python","args": ["/path/to/feedback_server.py"]}}}重启 Claude Desktop 后，聊天界面右上角会多出一个锤子图标，点开即可看到已注册的工具。直接在对话中提问"最近有哪些关于登录问题的反馈？"，Claude 会自动调用 search_feedback 工具，整个过程对用户来说几乎无感知。
方式二：用 FastMCP Client 编写代码连接
# client.pyimport asynciofrom fastmcp import Clientclient = Client("https://localhost:8000/mcp")async def main():async with client:# 调用工具result = await client.call_tool("search_feedback", {"keyword": "登录"})print(result)# 读取资源schema = await client.read_resource("data://schema")print(schema)asyncio.run(main())FastMCP Client 采用异步机制，必须在 async with 上下文环境中使用，这一点需要特别留意。
进阶功能：添加 Prompt 模板
MCP 还有一个非常实用的功能——Prompt（提示词模板），相当于为 AI 准备了一套操作指南，告诉它在面对特定任务时如何组织对话：
@mcp.promptdef analyze_negative(min_rating: int = 3) -> str:"""分析低评分反馈的提示词模板。"""return (f"请帮我分析评分低于 {min_rating} 的用户反馈。""步骤：1. 先用 get_stats 看整体情况；""2. 搜索评分低的反馈内容；""3. 归纳主要问题类别；""4. 给出改进建议。")Prompt 不是工具，它不直接执行操作。它的作用是给 AI 提供一份"操作手册"，告诉它在处理某类任务时应按什么顺序调用哪些工具。
踩坑记录
坑 1：同步函数会导致阻塞吗？
不会。FastMCP 3.x 默认将同步函数放入线程池中执行（run_in_thread=True），因此 sqlite3 这类同步库可以直接使用，无需改为 aiosqlite。多个请求同时到达也不会互相阻塞。
但如果你的函数涉及线程亲和性（例如 Windows 上的 COM 组件），需要设置 @mcp.tool(run_in_thread=False) 使其在事件循环线程中运行。
坑 2：*args 和 **kwargs 不可使用
FastMCP 需要完整的参数签名以生成 JSON Schema，因此工具函数不能包含 *args 或 **kwargs。所有参数必须明确声明类型。
这个限制一开始没留意，写了个 def query(sql: str, **params)，结果启动时直接报错。
坑 3：HTTP 传输的 URL 路径
使用 HTTP 传输时，MCP 端点的路径是 /mcp，而非根路径。Client 连接地址应写 https://localhost:8000/mcp，而不是 https://localhost:8000/。第一次连接时 404 了好一会儿才意识到问题所在。
坑 4：Resource URI 格式
Resource 的 URI 必须包含 scheme，例如 data://schema 或 file:///path。写成 schema 会报 invalid URI 错误。这个要求在官方文档中有说明，但很容易被忽略。
stdio vs HTTP：如何选择
| 场景 | 传输方式 | 原因 |
|---|---|---|
| Claude Desktop 本地连接 | stdio | 无网络开销，启动速度快 |
| 多个客户端共享一个 Server | HTTP | stdio 仅支持单客户端 |
| 远程部署 | HTTP | 必须通过网络通信 |
| 开发调试 | stdio | 简单便捷，无需管理端口 |
stdio 是默认值，适合本地使用场景。HTTP 采用 Streamable HTTP 协议（并非普通 REST API），支持 Server-Sent Events 实现流式响应。
生产环境补充建议
如果要部署到生产环境，还有几个要点需要注意：
认证机制：FastMCP 支持 OAuth 和 Token 验证。HTTP 传输时建议启用认证，避免未授权访问：
from fastmcp.server.auth import BearerTokenVerifiermcp = FastMCP("生产服务",auth=BearerTokenVerifier(token="your-secret-token"))超时控制：为工具设置超时时间，防止慢查询拖垮整个 Server：
@mcp.tool(timeout=10.0)# 10 秒超时def slow_query(sql: str) -> list[dict]:...中间件：FastMCP 3.x 支持 Middleware，可用于实现日志记录、限流、错误处理等横切关注点，与 Web 框架的中间件概念类似。
与直接写 function calling 相比，MCP 的优势在哪
之前用 OpenAI 的 function calling 也做过类似功能。对比下来，MCP 的优势主要体现在：
- 协议标准化：function calling 各家 API 格式不统一，MCP 是开放协议，编写一次 Server，所有支持 MCP 的客户端都能使用
- Server 独立部署：工具逻辑与 AI 应用解耦，修改数据库查询无需改动 AI 侧的代码
- 发现机制：客户端可以自动发现 Server 提供了哪些工具、资源和提示词，无需手动维护 schema 列表
- 传输灵活：同一套代码既可在本地运行（stdio），也可部署在远端（HTTP），无需修改业务逻辑
当然，MCP 并非万能方案。如果只使用 OpenAI 一家的 API，且工具逻辑非常简单，直接使用 function calling 反而更快捷。MCP 更适合工具种类多、客户端多、需要复用的复杂场景。
小结
FastMCP 几乎屏蔽了 MCP 协议的底层复杂度。编写一个 Python 函数加上一个装饰器，就是一个可用的 MCP 工具。对于希望为 AI 对接外部数据源的场景，这种方案比自己封装 API 再套一层 function calling 要省事得多。
文中所有代码均可直接运行。想要尝试的话，先安装 FastMCP，参照上面的示例修改数据库路径即可。
