将语言模型连接到结构化数据库,可使其从文本生成器变为数据分析师。虽然资源允许您向模型提供原始文本或日志,但对于受上下文窗口限制的大型数据集来说,它们通常不够用。然而,工具允许大型语言模型(LLM)使用结构化查询语言(SQL)请求特定数据片段。在本节中,我们将构建一个MCP服务器,该服务器向LLM公开一个SQLite数据库。这要求建立一个安全环境,模型可以在其中构建查询,服务器执行这些查询,并将结果作为JSON字符串返回。数据库工具的架构在编写代码之前,了解数据流很有帮助。与静态资源读取不同,工具调用涉及双向交互,其中客户端提供参数(SQL查询),服务器返回执行结果。digraph G { rankdir=TB; node [fontname="Arial", shape=box, style=filled, color="#dee2e6"]; edge [fontname="Arial", fontsize=10, color="#868e96"]; Client [label="MCP 客户端\n(Claude 桌面应用 / IDE)", fillcolor="#e9ecef"]; Server [label="MCP 服务器\n(FastMCP)", fillcolor="#4dabf7", fontcolor="white"]; Validation [label="输入验证\n(Pydantic)", fillcolor="#ced4da"]; Logic [label="执行逻辑\n(sqlite3)", fillcolor="#ffec99"]; Database [label="SQLite 文件\n(products.db)", fillcolor="#51cf66", fontcolor="white"]; Client -> Server [label="call_tool('query_db', {sql: '...'})"]; Server -> Validation [label="验证模式"]; Validation -> Logic [label="安全 SQL 字符串"]; Logic -> Database [label="执行查询"]; Database -> Logic [label="返回行"]; Logic -> Server [label="格式化 JSON"]; Server -> Client [label="ToolResult"]; }该图展现了请求的生命周期。服务器充当中间人,在逻辑层与物理数据库文件交互之前,验证工具参数。环境设置在本次练习中,我们将使用Python标准库中包含的sqlite3以及mcp SDK。我们将使用FastMCP类,它简化了JSON-RPC连接的初始设置和能力协商。首先,为本项目创建一个专用目录并初始化一个虚拟环境。mkdir mcp-db-tool cd mcp-db-tool python -m venv venv source venv/bin/activate # 在 Windows 上使用: venv\Scripts\activate pip install mcp步骤 1:填充数据库数据为了测试我们的工具,我们需要数据。创建一个名为setup_db.py的文件,以初始化一个模拟的电商库存。此脚本将创建一个表并插入示例记录。import sqlite3 def create_dummy_data(): conn = sqlite3.connect("inventory.db") cursor = conn.cursor() # 创建表 cursor.execute(""" CREATE TABLE IF NOT EXISTS products ( id INTEGER PRIMARY KEY, name TEXT NOT NULL, category TEXT, price REAL, stock_level INTEGER ) """) # 插入数据 products = [ ("Laptop Pro", "Electronics", 1299.99, 45), ("Wireless Mouse", "Electronics", 29.99, 150), ("Ergonomic Chair", "Furniture", 350.00, 20), ("Noise Cancelling Headphones", "Audio", 199.50, 60), ("USB-C Cable", "Accessories", 12.99, 300) ] cursor.executemany("INSERT INTO products (name, category, price, stock_level) VALUES (?, ?, ?, ?)", products) conn.commit() conn.close() print("数据库 'inventory.db' 已创建并包含示例数据。") if __name__ == "__main__": create_dummy_data()运行一次此脚本,会在您的目录中生成inventory.db文件。步骤 2:实现服务器现在我们创建实际的MCP服务器。创建一个名为server.py的文件。此文件将定义工具接口。在下面的代码中,请仔细查看query_products函数的文档字符串。MCP SDK使用此文档字符串来生成发送给LLM的工具定义中的描述字段。这段文本越详细,模型就越能更好地构建有效查询。from mcp.server.fastmcp import FastMCP import sqlite3 import json # 初始化服务器 mcp = FastMCP("Inventory Explorer") @mcp.tool() def query_products(sql_query: str) -> str: """ 对产品数据库执行SQL查询。 数据库中有一个名为'products'的表,包含以下列: id (整数), name (文本), category (文本), price (实数), stock_level (整数)。 只允许SELECT查询。 """ # 基本安全检查:强制只读 if not sql_query.strip().upper().startswith("SELECT"): raise ValueError("出于安全考虑,只允许SELECT查询。") try: # 如果URI支持,以只读模式连接到数据库, # 否则依赖于上面的逻辑检查。 conn = sqlite3.connect("inventory.db") cursor = conn.cursor() cursor.execute(sql_query) # 获取所有结果 rows = cursor.fetchall() # 获取列名以使结果具有自描述性 column_names = [description[0] for description in cursor.description] # 将列和行组合成字典列表 results = [dict(zip(column_names, row)) for row in rows] conn.close() # 将结果作为JSON字符串返回 return json.dumps(results, indent=2) except sqlite3.Error as e: return f"数据库错误:{str(e)}" except Exception as e: return f"执行错误:{str(e)}" if __name__ == "__main__": mcp.run()步骤 3:代码实现解析上述代码中有几处实现细节值得注意。用于验证的类型提示 参数sql_query: str不仅仅是为了可读性。MCP SDK使用Python类型提示来构建工具的JSON Schema。当客户端(LLM)调用此工具时,它知道必须提供一个字符串。如果我们为更复杂的参数(例如按特定价格范围过滤)使用了Pydantic模型,SDK会在函数运行之前强制执行这些限制。将错误处理作为上下文 注意,我们捕获异常并将其作为字符串返回,而不是让服务器崩溃。在MCP架构中,错误消息是有效的上下文。如果LLM编写了格式错误的SQL查询(例如,SELEECT * FROM products),返回“数据库错误:'SELEECT'附近:语法错误”可以帮助LLM识别其错误并在下一次对话中自行修正。安全考量 赋予LLM执行代码或SQL的能力本身就带有风险。在我们的示例中,我们添加了一个基本的startswith("SELECT")检查。在生产环境中,这还不够。您应该优先考虑:只读数据库用户: 创建一个只拥有SELECT权限的特定数据库用户。预定义查询: 不要接受原始SQL,而是创建get_products_by_category(category: str)这样的工具,其中SQL硬编码在您的Python函数中,LLM只提供参数。步骤 4:检查工具定义服务器启动时,会向客户端发送一个tools/list响应。根据我们的代码,生成的Schema将类似于下面的JSON结构。这就是LLM“看到”的,用以理解如何与您的数据库进行交互。{ "name": "query_products", "description": "对产品数据库执行SQL查询。数据库中有一个名为'products'的表...", "inputSchema": { "type": "object", "properties": { "sql_query": { "type": "string" } }, "required": ["sql_query"] } }测试工具为了测试此实现,您可以使用MCP Inspector或将其直接与客户端集成。如果您正在使用Claude桌面应用,您需要在配置文件中添加以下内容来注册服务器:"mcpServers": { "inventory": { "command": "uv", "args": [ "run", "server.py" ] } }连接后,您可以向模型提出自然语言问题,例如:“最贵的电子产品是什么?”“库存少于50件的商品有多少?”模型会将这些意图转换为SQL,调用query_products,并使用JSON响应来回答您的问题。工具与资源的区别您可能想知道我们为什么不为此使用资源。资源最适合静态数据,例如读取特定的日志文件(file://logs/error.txt)。如果我们为数据库使用资源,那么每次模型需要信息时,我们都必须将整个数据库内容转储到上下文窗口中。通过使用工具,我们允许模型执行数据检索。模型可以准确决定它需要哪些数据子集。这减少了token使用和延迟,并允许进行静态资源无法实现的操作,例如由数据库引擎而非LLM执行的聚合(计数、平均)。