Servidor MCP de base de datos

Un servidor de Protocolo de Contexto de Modelo (MCP) que proporciona herramientas para conectarse e interactuar con varios sistemas de bases de datos.

Características

Compatibilidad con múltiples bases de datos : conéctese a bases de datos SQLite, PostgreSQL, MySQL/MariaDB y SQL Server
Interfaz unificada : herramientas comunes para operaciones de bases de datos en todos los tipos de bases de datos compatibles
Extensiones específicas de la base de datos : cuando sea necesario, herramientas específicas para funciones específicas de la base de datos
Gestión de esquemas : crear, modificar y eliminar tablas e índices
Ejecución de consultas : ejecute consultas SQL sin procesar o utilice herramientas de consulta estructuradas
Soporte de transacciones : iniciar, confirmar y revertir transacciones

Instalación

Prerrequisitos

Python 3.8 o superior
Paquetes de Python necesarios (se instalan automáticamente con pip):
- SQLAlchemy
- Varios controladores de bases de datos, según las bases de datos que desee utilizar:
  - SQLite (incluido con Python)
  - PostgreSQL: psycopg2-binary
  - MySQL/MariaDB: mysql-connector-python
  - Servidor SQL: pyodbc

Instalación desde la fuente

# Clone the repository
git clone <repository-url>

# Install the package
pip install -e .

Configuración

El servidor se puede configurar utilizando variables de entorno, un archivo de configuración o proporcionando detalles de conexión en tiempo de ejecución.

Variables de entorno

DB_CONFIG_PATH : Ruta a un archivo de configuración JSON
DB_CONNECTIONS : una lista separada por comas de ID de conexión o una cadena JSON con detalles de conexión

Formato del archivo de configuración

{
  "connections": {
    "sqlite_conn": {
      "type": "sqlite",
      "db_path": "/path/to/database.db"
    },
    "postgres_conn": {
      "type": "postgres",
      "host": "localhost",
      "port": 5432,
      "database": "mydatabase",
      "user": "myuser",
      "password": "mypassword"
    }
  }
}

Uso

Ejecución del servidor

Como servidor MCP para Claude

# Run with default settings
python -m db_mcp_server

# Specify a configuration file
python -m db_mcp_server --config /path/to/config.json

# Set logging level
python -m db_mcp_server --log-level DEBUG

Como servidor web independiente (para cualquier LLM)

# Run as a web server
python -m db_mcp_server.web_server

# Specify host and port
python -m db_mcp_server.web_server --host 0.0.0.0 --port 8000

# Specify configuration file and logging level
python -m db_mcp_server.web_server --config /path/to/config.json --log-level DEBUG

Herramientas MCP disponibles

Gestión de conexiones

add_connection : Agregar una nueva conexión de base de datos
test_connection : prueba una conexión a la base de datos
list_connections : Lista todas las conexiones de la base de datos
remove_connection : Eliminar una conexión de base de datos

Ejecución de consultas

execute_query : Ejecutar una consulta SQL
get_records : Obtener registros de una tabla
insert_record : Insertar un registro en una tabla
update_record : Actualizar registros en una tabla
delete_record : Eliminar registros de una tabla

Gestión de esquemas

list_tables : enumera todas las tablas de una base de datos
get_table_schema : obtener el esquema de una tabla
create_table : Crea una nueva tabla
drop_table : Borrar una tabla
create_index : Crea un índice en una tabla
drop_index : Eliminar un índice
alter_table : Alterar una estructura de tabla

Gestión de transacciones

begin_transaction : Iniciar una transacción
commit_transaction : Confirmar una transacción
rollback_transaction : revertir una transacción

Ejemplos

Agregar una conexión

{
  "connection_id": "my_sqlite_db",
  "type": "sqlite",
  "db_path": "/path/to/database.db"
}

Ejecutar una consulta

{
  "connection_id": "my_sqlite_db",
  "query": "SELECT * FROM users WHERE age > ?",
  "params": [21]
}

Crear una tabla

{
  "connection_id": "my_sqlite_db",
  "table": "users",
  "columns": [
    {
      "name": "id",
      "type": "INTEGER",
      "primary_key": true,
      "nullable": false
    },
    {
      "name": "name",
      "type": "TEXT",
      "nullable": false
    },
    {
      "name": "email",
      "type": "TEXT",
      "nullable": true
    }
  ]
}

Insertar registros

{
  "connection_id": "my_sqlite_db",
  "table": "users",
  "data": {
    "name": "John Doe",
    "email": "john@example.com"
  }
}

Desarrollo

Ejecución de pruebas

# Run all tests
python -m unittest discover

# Run specific test file
python -m unittest tests.test_sqlite

Conexión desde otros LLM

Al ejecutarse como servidor web independiente, otros LLM (como Llama 3) pueden conectarse al servidor MCP de la base de datos mediante HTTP. El servidor expone los siguientes puntos finales:

Puntos finales

/list_tools - GET o POST: Devuelve una lista de todas las herramientas disponibles con sus descripciones y esquemas de entrada
/call_tool - POST: Ejecutar una herramienta de base de datos específica

Ejemplo: Llamar desde otro LLM

Para usar este servidor con otro LLM, haga que el LLM genere solicitudes HTTP al servidor. A continuación, se muestra un ejemplo de cómo podría estructurar la solicitud para un LLM como Llama 3:

You can interact with a database by making HTTP requests to a database service at http://localhost:8000. 
The service provides the following endpoints:

1. To get a list of available tools:
   Make a POST request to: http://localhost:8000/list_tools
   
2. To execute a database tool:
   Make a POST request to: http://localhost:8000/call_tool
   with a JSON body like:
   {
     "name": "tool_name",
     "arguments": {
       "param1": "value1",
       "param2": "value2"
     }
   }

For example, to execute a SQL query, you would make a request like:
POST http://localhost:8000/call_tool
Content-Type: application/json

{
  "name": "execute_query",
  "arguments": {
    "connection_id": "my_db",
    "query": "SELECT * FROM users"
  }
}

Código Python de muestra para la integración del cliente

import requests
import json

# Base URL of the database MCP server
BASE_URL = "http://localhost:8000"

# List available tools
def list_tools():
    response = requests.post(f"{BASE_URL}/list_tools")
    return response.json()

# Execute a database tool
def call_tool(tool_name, arguments):
    payload = {
        "name": tool_name,
        "arguments": arguments
    }
    response = requests.post(f"{BASE_URL}/call_tool", json=payload)
    return response.json()

# Example: List tables in a database
def list_tables(connection_id):
    return call_tool("list_tables", {"connection_id": connection_id})

# Example: Execute a SQL query
def execute_query(connection_id, query, params=None):
    return call_tool("execute_query", {
        "connection_id": connection_id,
        "query": query,
        "params": params
    })

# Example: Add a new connection
def add_connection(connection_id, db_type, **kwargs):
    args = {"connection_id": connection_id, "type": db_type}
    args.update(kwargs)
    return call_tool("add_connection", args)

Licencia

Licencia MIT