用Python构建一个MCP服务器,并将其连接到Claude Desktop和SQLite数据库。
用 Python 构建一个封装 SQLite 数据库的 MCP 服务器,将其注册到 Windows 版 Claude Desktop 中,让模型能够编写并执行自己的 SQL 查询。
来源: Stop Writing Custom APIs for Your AI Agents: Build an MCP Server in 5 Minutes
任何开发企业级智能体的人都知道编写连接代码的痛苦。需要一个数据库封装层,一个Jira工单集成层,还有一个Slack集成层。如果再加上十个数据库以及Claude、AutoGen和LangChain这五种大语言模型框架,设计工作就会因负担过重而崩溃。这就是N乘以M的集成问题:人工智能必须了解它所接触的每一个数据源的确切结构。
模型上下文协议(MCP)正是为解决这一问题而制定的开放标准。可以将其想象成人工智能领域的通用USB-C接口。数据库连接只需编写一次,通过统一的标准化服务器暴露出来,任何兼容的客户端都可以连接、发现可用的工具并加以使用。下面是一个用不到50行Python代码实现的本地MCP服务器示例,它能够连接到Claude Desktop,并由实时运行的SQLite数据库提供支持。
前置条件
官方的 Python mcp 包负责处理繁重的工作。它隐藏了 JSON-RPC 的细节,因此只需一个装饰器即可将工具暴露给模型。请先在您的虚拟环境中安装它:
pip install mcp
Claude Desktop 也需要安装在该机器上。
编写服务器代码
创建一个名为 mcp_server.py 的文件,导入所需库并初始化服务器。这里有一个重要细节:os 模块会将 SQLite 文件固定在与脚本相同的目录中。Claude Desktop 会在后台启动服务器,如果没有指定固定路径,就可能会出现权限错误。
import sqlite3
import json
import os
import sys
from mcp.server.mcpserver import MCPServer
# Initialize the MCP server
mcp = MCPServer(name="Enterprise_SQL_Agent")
# Force the database to be created in the exact same folder as this script
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
DB_PATH = os.path.join(BASE_DIR, "enterprise.db")
接下来是一个临时数据库,以便模型有数据可查询。日志记录会被输出到标准错误流(sys.stderr)。标准输出则是MCP与客户端通信的通道,若有无关内容输出到该通道会破坏数据完整性。
def setup_dummy_db():
"""Create a sample employee database for the demo"""
try:
conn = sqlite3.connect(DB_PATH)
cursor = conn.cursor()
cursor.execute('''CREATE TABLE IF NOT EXISTS employees
(id INTEGER PRIMARY KEY, name TEXT, role TEXT, salary INTEGER)''')
cursor.execute("DELETE FROM employees")
employees = [
("Alice", "Data Scientist", 120000),
("Bob", "DevOps Engineer", 115000),
("Charlie", "AI Researcher", 135000)
]
cursor.executemany("INSERT INTO employees (name, role, salary) VALUES (?, ?, ?)", employees)
conn.commit()
conn.close()
print("Database initialized successfully.", file=sys.stderr)
except Exception as e:
print(f"Database setup error: {e}", file=sys.stderr)
将数据库作为工具暴露
核心步骤是一个普通的 Python 函数,该函数执行 SQL 查询并返回结果行。加上 @mcp.tool() 装饰器后,服务器会读取文档字符串和类型提示,从而自动为该模型生成 JSON 模式。
@mcp.tool()
def query_employee_database(sql_query: str) -> str:
"""
Executes a SQL SELECT query against the enterprise.db database.
The database contains an 'employees' table with columns:
- id (INTEGER PRIMARY KEY)
- name (TEXT)
- role (TEXT)
- salary (INTEGER)
SECURITY: Only READ operations (SELECT) are permitted.
"""
# Safety Check: Block destructive SQL commands
dangerous_keywords = ["DROP", "DELETE", "UPDATE", "INSERT", "ALTER"]
if any(keyword in sql_query.upper() for keyword in dangerous_keywords):
return "Error: Only SELECT queries are authorized for this tool."
try:
conn = sqlite3.connect(DB_PATH)
cursor = conn.cursor()
cursor.execute(sql_query)
results = cursor.fetchall()
# Format the output as JSON so the LLM can read it cleanly
column_names = [description[0] for description in cursor.description]
formatted_results = [dict(zip(column_names, row)) for row in results]
conn.close()
return json.dumps(formatted_results, indent=2)
except Exception as e:
return f"Database error: {str(e)}"
一段简短的启动代码块便完成了该文件:
if __name__ == "__main__":
setup_dummy_db()
mcp.run()
连接 Claude Desktop
需要告知客户端服务器的位置。在 Windows 系统中,进入 %APPDATA%\Claude 文件夹,用编辑器打开 claude_desktop_config.json,然后添加一个 mcpServers 条目。
Windows 系统中一个常让人犯错的细节是:如果在 Conda 或 venv 等虚拟环境中运行程序,应提供该环境下 python.exe 的绝对路径,而非通用的全局 python 命令,且所有路径都需使用双反斜杠表示。
{
"mcpServers": {
"enterprise-sql": {
"command": "C:\\Users\\YourName\\.conda\\envs\\your_env\\python.exe",
"args": [
"D:\\Your\\Project\\Path\\mcp_server.py"
]
}
}
}
保存文件后,彻底重启 Claude Desktop。
验证功能
打开 Claude Desktop,查看聊天框右下角的锤子图标(工具)。工具 query_employee_database 应已注册在那里。一个不错的初步测试是询问员工表中收入最高的人是谁以及其职位是什么。
模型会读取工具描述,自行编写 SQL 语句(SELECT * FROM employees ORDER BY salary DESC LIMIT 1;),通过你的服务器执行该语句,然后将结果转化为通俗易懂的答案。无需编写自定义 API 层,架构也是动态发现的。
同样的模式也适用于 Slack、Jira、GitHub 或任何企业内部 API,而这正是改变企业构建智能系统成本结构的关键:每个系统只需被封装一次,所有兼容的助手都能使用它。