MCP Server 从0到1实战:手写一个让AI调用本地数据库的Tool工具
Agent能聊天了,但调不了我的本地数据
前段时间团队在做内部知识库的AI助手,Agent能联网搜索,能调公开API,但一碰到"帮我查一下上个月订单库的退款率"就哑了。数据在本地SQLite里,Agent够不着。当时想了几个方案——把数据库暴露成REST API?数据同步到云端?都不太对味。后来接触到MCP Server,简单说就是让AI能调你本地服务的标准化协议。本文从零开始手写一个MCP Server,注册几个Tool:一个查本地数据库,一个搜本地文件。代码都能直接跑,踩过的坑也一并写上。
MCP Server 一句话定位
先说结论:MCP Server 做的事就是把本地服务包装成AI能理解、能调用的Tool接口。结构大概是这样的:
AI Client ←→ MCP Client ←→ MCP Server ←→ Tool A: query_database
←→ Tool B: search_files
MCP Client 负责跟你的AI框架(LangChain、OpenAI SDK、自建Agent等)交互;MCP Server 负责跟你的本地资源交互。你只需要写Server端的Tool,Client端的通信协议MCP SDK已经封装好了。协议细节这里不展开,够写Server就行。
环境准备与传输方式选型
环境只需要装一个包就行:
# Python 3.10+
pip install mcp
Server启动后跟Client之间的通信方式有好几种。我当时调研了一圈,整理了一张选型表:
| 传输方式 | 适用场景 | 优点 | 缺点 |
|---|---|---|---|
| stdio | 本地CLI工具、同机部署 | 零配置,最稳定 | Client和Server必须在同一台机器 |
| SSE | 内网服务、远程AI Agent | 标准HTTP,可跨机器 | 需要Web服务容器 |
| 自定义传输 | 特殊协议、高安全要求 | 灵活度高 | 维护成本最高 |
我的经验是初期先用stdio起步,调试成本最低。跑通了再考虑要不要切SSE。这次代码就用stdio。
Step 1:初始化 Server 并注册第一个 Tool
先搭架子,让Server能正常启动。这段代码解决的问题是:确认环境没问题,Client能连上来。这样后面加正式功能的时候,基础链路已经是通的——如果这里跑不通,后面加再多功能都没用。
# server.py
from mcp.server.fastmcp import FastMCP
# 创建 MCP Server 实例,使用 stdio 传输
mcp = FastMCP("local-tools", transport="stdio")
@mcp.tool()
def health_check() -> str:
"""Check if the server is running and accessible."""
return "ok"
if __name__ == "__main__":
mcp.run()
运行后终端进入等待状态,表示Server已启动。你可以用MCP Client来连一下看效果:
# client_test.py
from mcp.client import StdioClient
async def main():
async with StdioClient(["python", "server.py"]) as client:
result = await client.call_tool("health_check", {})
print(result) # 输出: ok
如果输出 ok,就说明你的MCP环境通了——这时Agent已经可以通过Client跟本地服务通信了,虽然只是打了个招呼。不ok的话,常见问题是Python版本或mcp包没装好,先检查这两项。
翻车案例 1:Tool 参数描述写太简略
第一个Tool跑通后,我开始加真正有用的Tool,但很快就翻车了。看看这几版写法导致的差异:
| Tool 描述写法 | AI 实际行为 | 问题 |
|---|---|---|
| “查询数据库” | AI 传了 DROP TABLE orders |
AI 以为这是通用SQL接口,什么都能跑 |
| “执行SELECT查询,参数为完整SQL语句” | AI 只传 SELECT 语句 | 范围限制了,但偶尔传错表名 |
| “查询订单数据,可指定日期范围和状态” | AI 只传合理的 WHERE 条件 | 给了参数格式示例,AI照做 |
我的第一个版本是这样的:
@mcp.tool()
def query_database(sql: str) -> str:
"""查询数据库"""
# ...
结果AI传了什么?它传了 "SELECT * FROM users; DROP TABLE users;"。倒不是说AI有恶意,而是Tool的description对AI来说就是API文档——你写"查询数据库",AI就以为这是个执行任意SQL的接口。
根因分析: MCP协议的Tool描述会被直接注入到AI的system prompt里。AI根据description和参数名来"猜测"这个Tool的用途和边界。description越模糊,AI的猜测空间越大,翻车概率就越高。回过头来看确实是我想简单了——凭啥AI知道"查询数据库"就等于"只读查询"?
修正后的写法,把边界条件写在description第一句,参数给一个完整的示例值:
@mcp.tool()
def query_database(sql: str) -> str:
"""Execute SELECT queries on the local order database.
IMPORTANT: Only SELECT statements are supported.
INSERT, UPDATE, DELETE, DROP and other DDL/DML will be rejected.
Args:
sql: A complete SELECT SQL statement, e.g. "SELECT * FROM orders WHERE date > '2026-01-01'"
"""
import sqlite3
if not sql.strip().upper().startswith("SELECT"):
return json.dumps({"error": "Only SELECT queries are permitted"})
conn = sqlite3.connect("orders.db")
try:
cursor = conn.execute(sql)
rows = cursor.fetchall()
columns = [desc[0] for desc in cursor.description]
return json.dumps([dict(zip(columns, row)) for row in rows])
except Exception as e:
return json.dumps({"error": str(e)})
finally:
conn.close()
这样AI看一眼就知道该怎么传——实际调用时,AI会传类似 "SELECT order_id, amount FROM orders WHERE date > '2026-01-01'" 这样的SQL,不再触碰INSERT或DROP。后来我把这个规律总结成一个原则:Tool的描述要像给完全不了解你业务的人写API文档一样详细,不能留任何"常识性"的猜测空间。
Step 2:注册文件搜索 Tool
数据库之外,文件系统是另一个常见的本地资源。比如Agent问"帮我找一下上个月的周报文件",这时候就需要文件搜索能力。下面这个Tool接收目录名和文件通配符,返回匹配的文件路径列表,外加一些安全检查来防止不必要的风险。
import os
import glob
import json
@mcp.tool()
def search_files(directory: str, pattern: str = "*.md") -> str:
"""Search for files in the local documents directory.
Only searches within the /data/documents base directory.
Results are limited to protect system resources.
Args:
directory: Subdirectory name within /data/documents, e.g. "reports/2026"
pattern: Glob pattern, e.g. "*.md", "*.pdf", "weekly_*.xlsx"
"""
BASE = os.path.abspath("/data/documents")
target = os.path.abspath(os.path.join(BASE, directory))
# 防止目录穿越攻击
if not target.startswith(BASE):
return json.dumps({"error": "Access denied: directory outside allowed scope"})
if not os.path.exists(target):
return json.dumps({"error": f"Directory not found: {directory}"})
matches = glob.glob(os.path.join(target, pattern), recursive=True)
# 限制返回数量,防止AI循环调用把磁盘撑爆
limited = [os.path.relpath(p, BASE) for p in matches[:30]]
return json.dumps({"files": limited, "total": len(matches)})
运行后效果是这样的:
输入: search_files(directory="reports/2026", pattern="weekly_*.md")
输出: {"files": ["reports/2026/weekly_01.md", "reports/2026/weekly_02.md"], "total": 12}
这段代码加了两层防护:目录穿越检查(防止AI被注入 ../../etc/passwd 之类的路径)和结果数量限制。AI拿到文件列表后,可以对文件做后续操作——比如读取内容、总结摘要。不过这里得注意,glob 在大目录下可能比较慢,几十万文件的场景不太适合。
翻车案例 2:MCP 协议版本不一致
Step 1 和 Step 2 写完,我满心以为能跑通。结果Client连上来报了这个错:
Error: Unsupported protocol version. Server supports: 2024-11-05, Client supports: 2025-01-01
查了一圈发现MCP协议在2025年初有一次传输层规范更新。我装的最新版 mcp SDK默认用了新版协议,但Client那边引用的旧版本SDK还在用老的。两边的protocol version对不上,握手失败。
根因分析: MCP协议还在快速迭代中,当前的做法是每次连接握手时双方交换支持的版本号列表,取交集。如果Server和Client的SDK版本差了超过一个minor版本,交集大概率为空。这个问题在早期尤其频繁——我后来统计了一下,那段时间遇到版本不兼容的大概占了排查问题的将近一半。
修复办法其实很简单:Server和Client用同一个版本的 mcp 包。比较稳妥的做法是在项目里锁版本:
pip install mcp==1.2.0
然后在 requirements.txt 里写明依赖版本。后来我跟同事聊,他说等协议稳定了可能不用这么小心,但现阶段还是得看着。
整体验证
写完之后完整的启动流程就这几步:先启动 Server(终端进入等待状态),然后用 Client 或 MCP CLI 测试连接,最后依次调用 query_database 和 search_files 看能不能拿到正确结果。如果两个Tool都返回了预期的数据,链路就是通的。
验证通过后,你就有了一个跑在本地的MCP Server,AI Agent可以通过它查询数据库和搜索文件。整个上手过程前前后后差不多花了一天——写代码占了大半,排查协议版本问题用了一小半。
局限性与复用建议
自己写MCP Server不是万能的,有几个场景可能需要换个思路:
- 只有少量工具且在单机运行 —— 自己写Server最合适,成本最低
- 工具多了需要外部访问 —— 考虑用MCP官方市场现成的Server,或者用社区工具包装
- 对安全要求极高(金融/医疗) —— 自己写并且严格审计每个Tool的权限边界
- 团队前端为主不太熟Python —— MCP也有Node.js SDK,可以换Node写
另外还有一个容易被忽略的点:MCP Server本身不处理认证和授权。如果你的本地服务需要鉴权,你得自己在Tool实现里加,协议层不管这个。
回头看,踩坑的核心就两个:Tool描述要像给不懂业务的人写API文档一样详细,依赖版本一定锁死。前一个是设计问题,后一个是工程问题。做MCP Server这两件事都绕不开。
更多推荐



所有评论(0)