摘要： 手把手教你用Python写一个MCP Server，让Claude直接连SQLite数据库查数据，全程一个Demo，小白可跟做。
老板要数据，你还在手写SQL？
周五快下班，老板群里@你："查下上个月Pro用户的复购率。"
你打开Navicat，连数据库，写SQL——JOIN、GROUP BY、HAVING。查完导出Excel，整理格式，发群里。一看时间，40分钟没了。
如果你能直接说一句话呢？"帮我查上个月Pro用户的复购率。"10秒，结果出来了，还自动帮你生成报表。
很多人觉得AI只会聊天，干不了实事。但2026年有个叫MCP的协议正在改变这件事——它让AI从嘴炮选手变成了能动手的实习生。
这篇文章，带你从零写一个MCP Server，让Claude直接查你的SQLite数据库。全程一个Demo，每一步都有代码，小白也能跟着做完。
动手前的5分钟
打开终端，确认Python版本不低于3.9：
python3 --version
装一个包：
pip install mcp
验证一下：
python3 -c "import mcp; print('OK')"
看到OK，环境没问题。接下来准备Demo数据库，新建init_db.py，粘贴以下代码：
import sqlite3conn = sqlite3.connect("demo.db")c = conn.cursor()c.execute('''CREATE TABLE IF NOT EXISTS users ( id INTEGER PRIMARY KEY, name TEXT, email TEXT, plan TEXT, created_at TEXT)''')c.execute('''CREATE TABLE IF NOT EXISTS orders ( id INTEGER PRIMARY KEY, user_id INTEGER, amount REAL, status TEXT, created_at TEXT)''')c.execute("SELECT COUNT(*) FROM users")if c.fetchone()[0] == 0: users = [ (1, "张三", "zhangsan@mail.com", "pro", "2025-01-15"), (2, "李四", "lisi@mail.com", "free", "2025-03-20"), (3, "王五", "wangwu@mail.com", "pro", "2025-02-10"), ] orders = [ (1, 1, 99.0, "completed", "2025-06-01"), (2, 1, 199.0, "completed", "2025-07-15"), (3, 2, 49.0, "pending", "2025-08-01"), (4, 3, 99.0, "completed", "2025-06-20"), (5, 3, 299.0, "refunded", "2025-07-01"), ] c.executemany("INSERT INTO users VALUES (?,?,?,?,?)", users) c.executemany("INSERT INTO orders VALUES (?,?,?,?,?)", orders)conn.commit()conn.close()print("数据库初始化完成")
运行：
python3 init_db.py
看到"数据库初始化完成"，Demo数据库就准备好了——两张表，3个用户，5条订单。复制粘贴运行就行，别跳过，后面所有实操都基于这个库。
写你的第一个MCP Server
先写最简版，10行代码，让你感受MCP是怎么回事。新建server.py：
from mcp.server.fastmcp import FastMCPmcp = FastMCP("demo-server")@mcp.tool()def list_tables() -> str: """列出数据库中所有表名""" import sqlite3 conn = sqlite3.connect("demo.db") cursor = conn.execute( "SELECT name FROM sqlite_master WHERE type='table'") tables = [row[0] for row in cursor.fetchall()] conn.close() return f"数据库中的表: {', '.join(tables)}"if __name__ == "__main__": mcp.run()
保存，运行：
python3 server.py
没报错就成功了。@mcp.tool()是核心，它把普通Python函数注册成AI可调用的工具。docstring自动变成工具描述，类型注解自动生成参数schema，你什么都不用额外写。
想验证工具是否注册成功？用MCP Inspector。另开一个终端：
npx @modelcontextprotocol/inspector python3 server.py
浏览器打开http://localhost:5173，你会看到可视化界面，里面列出了list_tables工具，还能直接点击测试。
看到这里你可能想：就这？别急，接下来才是重头戏。现在写完整版，注册4个工具。
新建db_server.py：
import sqlite3from mcp.server.fastmcp import FastMCPmcp = FastMCP("database-server")def get_db(): return sqlite3.connect("demo.db")@mcp.tool()def list_tables() -> str: """列出数据库中所有表名""" conn = get_db() cursor = conn.execute( "SELECT name FROM sqlite_master WHERE type='table'") tables = [row[0] for row in cursor.fetchall()] conn.close() return f"数据库中的表: {', '.join(tables)}"@mcp.tool()def describe_table(table_name: str) -> str: """查看指定表的结构 Args: table_name: 表名 """ conn = get_db() try: cursor = conn.execute( f"PRAGMA table_info({table_name})") columns = cursor.fetchall() if not columns: return f"表 {table_name} 不存在" result = f"表 {table_name} 结构:\n" for col in columns: result += f" {col[1]} ({col[2]})\n" return result finally: conn.close()@mcp.tool()def execute_query(sql: str) -> str: """执行SQL查询（仅允许SELECT） Args: sql: SELECT查询语句 """ normalized = sql.strip().upper() if not normalized.startswith("SELECT"): return "安全限制：只允许SELECT查询" dangerous = ["DROP","DELETE","UPDATE","INSERT","ALTER"] if any(kw in normalized for kw in dangerous): return "安全限制：检测到危险操作" conn = get_db() try: cursor = conn.execute(sql) rows = cursor.fetchall() columns = [desc[0] for desc in cursor.description] if not rows: return "查询结果为空" result = " | ".join(columns) + "\n" result += "-" * len(result) + "\n" for row in rows[:50]: result += " | ".join(str(v) for v in row) + "\n" if len(rows) > 50: result += f"... 共{len(rows)}行，已截断" return result except Exception as e: return f"查询错误: {str(e)}" finally: conn.close()@mcp.tool()def get_user_summary(user_id: int) -> str: """获取用户摘要，含订单统计 Args: user_id: 用户ID """ conn = get_db() try: user = conn.execute( "SELECT * FROM users WHERE id=?", (user_id,) ).fetchone() if not user: return f"用户ID {user_id} 不存在" stats = conn.execute(""" SELECT COUNT(*) as total, SUM(CASE WHEN status='completed' THEN amount ELSE 0 END) as spent, SUM(CASE WHEN status='pending' THEN 1 ELSE 0 END) as pending FROM orders WHERE user_id=? """, (user_id,)).fetchone() return f"""用户: {user[1]} (ID: {user[0]})邮箱: {user[2]}套餐: {user[3]}注册: {user[4]}订单数: {stats[0]}已消费: ¥{stats[1]:.2f}待处理: {stats[2]}""" finally: conn.close()if __name__ == "__main__": mcp.run()
4个工具各有分工：list_tables看库里有啥表，describe_table看表结构，execute_query执行任意SELECT，get_user_summary一键拉用户画像。
注意execute_query里的安全检查。我做了两层拦截——第一层只允许SELECT开头，第二层扫描危险关键词。别小看这个设计，我第一次写的时候没加，AI帮我生成了一条DROP TABLE，差点把测试库删了。从那以后再也不敢省安全检查。
接入Claude，让AI干活
Server写好了，怎么让Claude用起来？两种方案，一种零代码，一种写几行Python。
方案一：Claude Desktop接入
找到配置文件。macOS路径：~/Library/Application Support/Claude/claude_desktop_config.json，Windows路径：%APPDATA%\Claude\claude_desktop_config.json。
打开，改成这样：
{ "mcpServers": { "database": { "command": "python3", "args": ["/你的路径/db_server.py"] } }}
把路径换成你实际的，保存，重启Claude Desktop。打开Claude，直接说："帮我查一下数据库里有什么表。"
你会看到Claude自动调用了list_tables，返回了"users, orders"。它不是在猜，它真的连上了你的数据库，真的执行了查询。
再试一句："Pro用户一共消费了多少？"Claude自动调用execute_query，生成一条JOIN查询，执行后告诉你结果。你不需要写任何SQL。
我第一次跑通的时候，盯着屏幕愣了大概10秒。那种感觉怎么说呢——就像你一直开手动挡，突然发现它有自动挡。
方案二：Python接入GPT-4
想在代码里集成？用Python客户端，新建client.py：
import json, asynciofrom openai import OpenAIfrom mcp import ClientSession, StdioServerParametersfrom mcp.client.stdio import stdio_clientclient = OpenAI()async def run(): server_params = StdioServerParameters( command="python3", args=["db_server.py"]) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() tools_result = await session.list_tools() openai_tools = [{ "type": "function", "function": { "name": t.name, "description": t.description, "parameters": t.inputSchema } } for t in tools_result.tools] print(f"已加载 {len(openai_tools)} 个工具") messages = [] while True: q = input("\n你: ") if q in ('quit', 'q'): break messages.append( {"role": "user", "content": q}) resp = client.chat.completions.create( model="gpt-4o", messages=messages, tools=openai_tools, tool_choice="auto") msg = resp.choices[0].message if msg.tool_calls: messages.append(msg) for tc in msg.tool_calls: args = json.loads( tc.function.arguments) print(f" 调用: " f"{tc.function.name}({args})") result = await session.call_tool( tc.function.name, args) messages.append({ "role": "tool", "tool_call_id": tc.id, "content": result.content[0].text}) final = client.chat.completions.create( model="gpt-4o", messages=messages, tools=openai_tools) print(f"\nAI: " f"{final.choices[0].message.content}") messages.append({ "role": "assistant", "content": final.choices[0].message.content}) else: print(f"\nAI: {msg.content}") messages.append({ "role": "assistant", "content": msg.content})asyncio.run(run())
运行python3 client.py，你就能在终端里跟"能查数据库的GPT-4"对话了。核心逻辑就三步：连MCP Server拿工具列表，把工具转成OpenAI格式，在对话循环里自动调用。
实测效果：
你: 查一下Pro用户有多少个，总消费多少 调用: execute_query({"sql": "SELECT COUNT(*), SUM(amount) FROM users u JOIN orders o ON u.id=o.user_id WHERE u.plan='pro' AND o.status='completed'"})AI: Pro用户共2人，已完成订单总消费397元。
AI自己决定了调哪个工具、传什么参数、怎么拼SQL。你只说了一句话。
我踩过的坑，你不用再踩
坑一：SDK没装就跑代码。 报错信息是一堆ModuleNotFoundError，新手看了会懵。解决很简单：pip install mcp。但很多人会忘。
坑二：配置路径写错。 macOS和Windows路径不一样，Windows用户尤其容易搞混。最靠谱的办法：在Claude Desktop里按Cmd+Shift+P（Mac）或Ctrl+Shift+P（Windows），搜"config"，直接打开配置文件。
坑三：没加SQL安全检查。 我第一次写的execute_query什么SQL都放行，结果AI生成了DROP TABLE users。还好连的是测试库，从那以后加了双层拦截——只允许SELECT，再扫描危险关键词。生产环境更严格，用只读数据库账号连接。
安全红线必须牢记：
- 敏感字段（密码、token）不要返回 工具设计也有讲究。工具名要一眼看懂，
fn(a, b)这种命名是灾难，query_user_by_name才合格。docstring要写清楚用途、参数含义、返回值格式，AI靠这些描述判断调用时机，描述模糊它就会乱调。每个工具只做一件事，别搞万能查询工具包揽一切，拆成三个，AI才能精准选择。
总结：AI从聊天进化到干活
MCP让AI从只会说话，进化到能干活。这不是技术升级，是AI应用范式的转变——从chatbot到agent，MCP是基础设施。
什么时候用MCP，什么时候用Function Calling？我的判断很简单：工具少于5个，只用一个模型，Function Calling够用。多人协作、多模型、工具多，上MCP。想接入Claude Desktop，必须用MCP。
MCP生态已经挺丰富了，官方有filesystem、sqlite、github、postgres等，社区有obsidian、slack、brave-search等，配置方式都一样——写进JSON配置文件就行。
我建议你的下一步：基于本文Demo，写一个连接你公司内部API的MCP Server。比如连上CRM系统，让AI直接查客户数据。或者连上监控平台，让AI直接拉报警记录。一旦用上，就回不去了。
