SQL MCP Server (sql-mcp-server)

MCP SQL Server

FastMCP服务器，通过对话式AI接口提供SQL数据库交互工具。

概述

该项目创建了一个服务器，通过会话式AI接口公开MS SQL server操作。它使用FastMCP框架提供查询和操作SQL数据的工具，允许用户使用自然语言与数据库交互。

特性

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

需求

• Python 3.7+
• pyodbc
• 异步IO
• FastMCP框架
• Microsoft SQL Server
• SQL Server的ODBC驱动程序17

安装

1. 安装Python依赖项：

pip install pyodbc asyncio fastmcp

1. 确保已安装Microsoft SQL Server和用于SQL Server的ODBC驱动程序17。

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

# 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数据库的连接。

可用工具

查询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_驱动程序

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

list_odbc_drivers() -> str

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

数据库信息

获取有关连接的数据库的一般信息。

database_info() -> str

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

建筑

服务器使用异步架构来避免阻塞操作：

1. 生命周期管理：The app_lifespan 上下文管理器处理数据库连接的建立和断开。

1. 非阻塞操作：数据库操作在单独的线程中运行，使用 asyncio.get_event_loop().run_in_executor() 以防止阻塞主事件循环。

1. 错误处理：所有操作都包括全面的错误处理和有用的错误消息。

错误处理

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

• 数据库连接失败
• 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驱动程序错误：确认已安装SQL Server的ODBC驱动程序17
3. 权限错误：检查运行应用程序的Windows用户是否具有适当的SQL Server权限

许可证

\[您的许可证信息\]

联系

\[您的联系方式\]