MCP SQL Server

MCP SQL Server

一个 FastMCP 服务，通过对话式 AI 界面提供 SQL 数据库交互工具。

概述

本项目创建了一个服务，通过对话式 AI 界面提供 MS SQL Server 操作。它使用 FastMCP 框架提供查询和处理 SQL 数据的工具，让用户可以通过自然语言与数据库交互。

功能

• 执行 SQL 查询并查看结果
• 列出数据库中的可用表
• 查看表结构及列信息
• 执行非查询操作（INSERT、UPDATE、DELETE）
• 列出系统中可用的 ODBC 驱动程序
• 查看数据库信息和服务详细信息

要求

• Python 3.7+
• pyodbc
• asyncio
• FastMCP 框架
• Microsoft SQL Server
• ODBC Driver 17 for SQL Server

安装

1. 安装 Python 依赖项：

pip install pyodbc asyncio fastmcp

2. 确保已安装 Microsoft SQL Server 和 ODBC Driver 17 for SQL Server。

3. 在脚本中配置连接设置：

# Connection parameters
SERVER = "server\\instance"  # Change to your SQL Server instance
DATABASE = "db_name"              # Change to your database name

使用方法

运行服务：

python mcp_sql_server.py

服务将初始化并连接到指定的 SQL Server 数据库。

可用工具

query_sql

执行 SQL 查询并返回结果。

query_sql(query: str = None) -> str

• 如果未提供查询语句，则默认使用 SELECT * FROM [dbo].[Table_1]
• 将查询结果作为格式化字符串返回

list_tables

列出数据库中的所有表。

list_tables() -> str

• 以字符串形式返回表名列表

describe_table

获取特定表的结构。

describe_table(table_name: str) -> str

• table_name：要查看结构的表名
• 返回列信息，包括列名和数据类型

execute_nonquery

执行 INSERT、UPDATE、DELETE 或其他非查询 SQL 语句。

execute_nonquery(sql: str) -> str

• sql：要执行的 SQL 语句
• 返回操作结果，包括受影响的行数
• 自动处理事务（提交/回滚）

list_odbc_drivers

列出系统中所有可用的 ODBC 驱动程序。

list_odbc_drivers() -> str

• 返回以逗号分隔的已安装 ODBC 驱动程序列表

database_info

获取已连接数据库的常规信息。

database_info() -> str

• 返回服务名称、数据库名称、SQL Server 版本、当前服务时间和表数量

架构

服务采用异步架构，以避免阻塞操作：

1. 生命周期管理：app_lifespan 上下文管理器负责建立和关闭数据库连接。

2. 非阻塞操作：数据库操作通过 asyncio.get_event_loop().run_in_executor() 在线程中运行，以防止阻塞主事件循环。

3. 错误处理：所有操作都包含完善的错误处理，并提供有用的错误消息。

错误处理

服务会处理各种错误情况：

• 数据库连接失败
• SQL 查询语法错误
• 表不存在错误
• 权限相关问题

所有错误都会记录日志，并向客户端返回相应的错误消息。

自定义

要添加新的数据库工具或修改现有工具，请参照现有工具采用的模式：

@mcp.tool()
async def your_new_tool(ctx: Context, param1: str) -> str:
    """Documentation for your tool"""
    try:
        conn = ctx.request_context.lifespan_context["conn"]
        
        if conn is None:
            return "Database connection is not available."
            
        def your_db_operation():
            # Your database operations here
            pass
            
        loop = asyncio.get_event_loop()
        result = await loop.run_in_executor(None, your_db_operation)
        
        # Process and return results
        return "Your result"
    except Exception as e:
        return f"Error: {str(e)}"

安全注意事项

• 服务使用 Windows 身份验证（"Trusted_Connection=yes"）
• 建议为 SQL 查询实现输入验证，以防止 SQL 注入
• 按照最小权限原则限制数据库用户权限

故障排除

常见问题：

1. 连接错误：验证 SQL Server 实例名称，并确保实例正在运行
2. ODBC 驱动程序错误：确认已安装 ODBC Driver 17 for SQL Server
3. 权限错误：检查运行应用程序的 Windows 用户是否具有适当的 SQL Server 权限

许可证

[您的许可证信息]

联系方式

[您的联系信息]