MCP SQL Server

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

概述

本项目创建一个服务器，通过对话式 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. 在脚本中配置连接设置：

# 连接参数
SERVER = "server\\instance"  # 更改为您的 SQL Server 实例
DATABASE = "db_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 权限

许可证

[您的许可证信息]

联系方式

[您的联系方式]