Servidor SQL MCP

Un servidor FastMCP que proporciona herramientas de interacción con bases de datos SQL a través de una interfaz de inteligencia artificial conversacional.

Descripción general

Este proyecto crea un servidor que expone las operaciones de MS SQL Server mediante una interfaz de IA conversacional. Utiliza el framework FastMCP para proporcionar herramientas de consulta y manipulación de datos SQL, permitiendo a los usuarios interactuar con bases de datos mediante lenguaje natural.

Características

Ejecutar consultas SQL y ver resultados
Lista de tablas disponibles en la base de datos
Describe la estructura de la tabla con información de columnas.
Ejecutar operaciones que no sean de consulta (INSERTAR, ACTUALIZAR, ELIMINAR)
Enumere los controladores ODBC disponibles en el sistema
Ver información de la base de datos y detalles del servidor

Requisitos

Python 3.7+
pyodbc
asincronía
Marco FastMCP
Microsoft SQL Server
Controlador ODBC 17 para SQL Server

Instalación

Instalar dependencias de Python:

pip install pyodbc asyncio fastmcp

Asegúrese de tener instalado Microsoft SQL Server y el controlador ODBC 17 para SQL Server.
Configure los ajustes de conexión en el script:

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

Uso

Ejecutar el servidor:

python mcp_sql_server.py

El servidor se inicializará y establecerá una conexión con la base de datos SQL Server especificada.

Herramientas disponibles

consulta_sql

Ejecutar una consulta SQL y devolver los resultados.

query_sql(query: str = None) -> str

Si no se proporciona ninguna consulta, el valor predeterminado es SELECT * FROM [dbo].[Table_1]
Devuelve los resultados de la consulta como una cadena formateada

listas_de_tablas

Enumere todas las tablas disponibles en la base de datos.

list_tables() -> str

Devuelve una lista de nombres de tablas como una cadena

describe_tabla

Obtener la estructura de una tabla específica.

describe_table(table_name: str) -> str

table_name : Nombre de la tabla a describir
Devuelve información de la columna, incluidos nombres y tipos de datos.

ejecutar_no consulta

Ejecute INSERT, UPDATE, DELETE u otras sentencias SQL que no sean de consulta.

execute_nonquery(sql: str) -> str

sql : La sentencia SQL a ejecutar
Devuelve los resultados de la operación, incluido el número de filas afectadas
Maneja automáticamente transacciones (confirmación/reversión)

lista_de_controladores_ODBC

Enumere todos los controladores ODBC disponibles en el sistema.

list_odbc_drivers() -> str

Devuelve una lista separada por comas de controladores ODBC instalados

información de la base de datos

Obtenga información general sobre la base de datos conectada.

database_info() -> str

Devuelve el nombre del servidor, el nombre de la base de datos, la versión de SQL Server, la hora actual del servidor y el recuento de tablas

Arquitectura

El servidor utiliza una arquitectura asincrónica para evitar operaciones de bloqueo:

Gestión del ciclo de vida : el administrador de contexto app_lifespan maneja la configuración y el desmontaje de la conexión a la base de datos.
Operaciones sin bloqueo : las operaciones de base de datos se ejecutan en un hilo separado utilizando asyncio.get_event_loop().run_in_executor() para evitar el bloqueo del bucle de eventos principal.
Manejo de errores : todas las operaciones incluyen un manejo de errores integral con mensajes de error útiles.

Manejo de errores

El servidor maneja varias condiciones de error:

Fallos de conexión a la base de datos
Errores de sintaxis de consultas SQL
Errores de tabla no encontrada
Problemas relacionados con los permisos

Se registran todos los errores y se devuelven los mensajes de error correspondientes al cliente.

Personalización

Para agregar nuevas herramientas de base de datos o modificar las existentes, siga el patrón utilizado en las herramientas existentes:

@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)}"

Consideraciones de seguridad

El servidor utiliza la autenticación de Windows ("Trusted_Connection=yes")
Considere implementar la validación de entrada para consultas SQL para evitar la inyección de SQL
Restringir los permisos de usuario de la base de datos según el principio del mínimo privilegio

Solución de problemas

Problemas comunes:

Errores de conexión : Verifique el nombre de la instancia de SQL Server y asegúrese de que se esté ejecutando
Errores del controlador ODBC : Confirme que el controlador ODBC 17 para SQL Server esté instalado
Errores de permisos : compruebe que el usuario de Windows que ejecuta la aplicación tenga los permisos adecuados de SQL Server

Licencia

[Información de su licencia]

Contacto

[Su información de contacto]

MCP SQL Server

Integrations