Database MCP Server

A Model Context Protocol (MCP) server that provides tools for connecting to and interacting with various database systems.

Features

Multi-Database Support: Connect to SQLite, PostgreSQL, MySQL/MariaDB, and SQL Server databases
Unified Interface: Common tools for database operations across all supported database types
Database-Specific Extensions: Where needed, specific tools for database-specific features
Schema Management: Create, alter, and drop tables and indexes
Query Execution: Execute raw SQL queries or use structured query tools
Transaction Support: Begin, commit, and rollback transactions

Installation

Prerequisites

Python 3.8 or higher
Required Python packages (installed automatically with pip):
- SQLAlchemy
- Various database drivers, depending on which databases you want to use:
  - SQLite (included with Python)
  - PostgreSQL: psycopg2-binary
  - MySQL/MariaDB: mysql-connector-python
  - SQL Server: pyodbc

Installing from Source

# Clone the repository
git clone <repository-url>

# Install the package
pip install -e .

Configuration

The server can be configured using environment variables, a configuration file, or by providing connection details at runtime.

Environment Variables

DB_CONFIG_PATH: Path to a JSON configuration file
DB_CONNECTIONS: A comma-separated list of connection IDs or a JSON string with connection details

Configuration File Format

{
  "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"
    }
  }
}

Usage

Running the Server

As an MCP Server for 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

As a Standalone Web Server (for any 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

Available MCP Tools

Connection Management

add_connection: Add a new database connection
test_connection: Test a database connection
list_connections: List all database connections
remove_connection: Remove a database connection

Query Execution

execute_query: Execute a SQL query
get_records: Get records from a table
insert_record: Insert a record into a table
update_record: Update records in a table
delete_record: Delete records from a table

Schema Management

list_tables: List all tables in a database
get_table_schema: Get the schema for a table
create_table: Create a new table
drop_table: Drop a table
create_index: Create an index on a table
drop_index: Drop an index
alter_table: Alter a table structure

Transaction Management

begin_transaction: Begin a transaction
commit_transaction: Commit a transaction
rollback_transaction: Rollback a transaction

Examples

Add a Connection

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

Execute a Query

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

Create a Table

{
  "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
    }
  ]
}

Insert Records

{
  "connection_id": "my_sqlite_db",
  "table": "users",
  "data": {
    "name": "John Doe",
    "email": "[email protected]"
  }
}

Development

Running Tests

# Run all tests
python -m unittest discover

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

Connecting from Other LLMs

When running as a standalone web server, other LLMs (like Llama 3) can connect to the database MCP server via HTTP. The server exposes the following endpoints:

Endpoints

/list_tools - GET or POST: Returns a list of all available tools with their descriptions and input schemas
/call_tool - POST: Execute a specific database tool

Example: Calling from Another LLM

To use this server with another LLM, have the LLM generate HTTP requests to the server. Here's an example of how you could structure the prompt for an LLM like 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"
  }
}

Sample Python Code for Client Integration

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)

License

MIT License