top of page
Search

Building a Database MCP Server: The Ultimate Code Walkthrough - Part 4

Updated: Jul 7

Prerequisite: This is a continuation of the blog Part 2: Building a Database MCP Server: The Ultimate Code Walkthrough - Part 3

ree

The Connection Traffic Controller - list_database_connections 

@mcp.tool()
async def list_database_connections() -> str:
    """List all active database connections"""
    connections = {}
    
    for key, manager in connection_pool.items():
        db_type, db_path = key.split(':', 1)
        connections[key] = {
            'database_path': db_path,
            'database_type': db_type,
            'is_sample_database': is_sample_database(db_path),
            'is_connected': manager.is_connected(),
            'connection_initialized': manager.is_initialized
        }
    
    return json.dumps({
        'status': 'success',
        'active_connections': connections,
        'total_connections': len(connections),
        'sample_database_path': SAMPLE_DB_PATH
    }, indent=2)

The Connection Pool Inspector! 🏊‍♂️

This function is like having a dashboard that shows all your open database connections with their health status and details.


Step 1: Connection Dictionary Initialization

    connections = {}

The Results Container:

  • Empty dictionary: Will be populated with connection details

  • Key-value structure: Each connection gets a unique entry

  • Flexible storage: Can handle any number of active connections


Step 2: Connection Pool Iteration

    for key, manager in connection_pool.items():

The Pool Walker:

  • connection_pool.items(): Gets both keys and manager objects

  • Key format: "sqlite:/path/to/db.db" or "postgresql:database_name"

  • Manager object: Our DatabaseManager instance with connection details


Step 3: Connection Key Parsing

        db_type, db_path = key.split(':', 1)

The Key Decoder:

  • split(':', 1): Splits only on the FIRST colon

  • Why limit to 1? Database paths might contain colons (Windows: "C:/path/db.db")

  • Result examples:

    • "sqlite:/data/app.db" → db_type="sqlite", db_path="/data/app.db"

    • "postgresql:host:port/dbname" → db_type="postgresql", db_path="host:port/dbname"


Step 4: Connection Health Assessment

        connections[key] = {
            'database_path': db_path,
            'database_type': db_type,
            'is_sample_database': is_sample_database(db_path),
            'is_connected': manager.is_connected(),
            'connection_initialized': manager.is_initialized
        }
The Health Check Compilation:

  • database_path: The actual file path or connection identifier

  • database_type: "sqlite" or "postgresql"

  • is_sample_database(db_path): Boolean flag for sample database detection

  • manager.is_connected(): Live health check - actually tests the connection!

  • manager.is_initialized: Whether the connection was properly set up


Step 5: Summary Report Generation

    return json.dumps({
        'status': 'success',
        'active_connections': connections,
        'total_connections': len(connections),
        'sample_database_path': SAMPLE_DB_PATH
    }, indent=2)

The Dashboard Assembly:

  • active_connections: Complete dictionary of all connection details

  • total_connections: Quick count for monitoring

  • sample_database_path: Reference path for the sample database

  • Consistent formatting: Same JSON structure as other tools


Output Example:

{
  "status": "success",
  "active_connections": {
    "sqlite:./data/sample_database.db": {
      "database_path": "./data/sample_database.db",
      "database_type": "sqlite",
      "is_sample_database": true,
      "is_connected": true,
      "connection_initialized": true
    },
    "sqlite:./production/app.db": {
      "database_path": "./production/app.db", 
      "database_type": "sqlite",
      "is_sample_database": false,
      "is_connected": true,
      "connection_initialized": true
    },

    "postgresql:my_enterprise_db": {
      "database_path": "my_enterprise_db",
      "database_type": "postgresql", 
      "is_sample_database": false,
      "is_connected": false,
      "connection_initialized": true
    }
  },

  "total_connections": 3,
  "sample_database_path": "./data/sample_database.db"
}

Why monitor connections? Like keeping track of open apps on your phone - you want to know what's using resources and whether everything is healthy!


The Cleanup Specialist! 🧹 - close_database_connection

@mcp.tool()
async def close_database_connection(db_path: str, db_type: str = "sqlite") -> str:
    """Close a specific database connection"""
    connection_key = f"{db_type}:{db_path}"
    
    if connection_key in connection_pool:
        manager = connection_pool[connection_key]
        manager.close()
        del connection_pool[connection_key]
        
        return json.dumps({
            'status': 'success',
            'message': f'Closed connection to {db_path}',
            'database_path': db_path,
            'database_type': db_type
        }, indent=2)
    else:
        return json.dumps({
            'status': 'info',
            'message': f'No active connection found for {db_path}',
            'database_path': db_path,
            'database_type': db_type
        }, indent=2)

This function is like having a professional janitor who knows exactly how to clean up database connections safely and thoroughly.


Step 1: Connection Key Reconstruction

connection_key = f"{db_type}:{db_path}"

The Key Builder:

  • Format matching: Creates the same key format used throughout the system

  • Consistency: Must exactly match how keys are stored in connection_pool

  • Examples:

    • "sqlite:./data/app.db"

    • "postgresql:enterprise_database"


Step 2: Connection Existence Check

if connection_key in connection_pool:

The "Is It There?" Verification:

  • Dictionary lookup: Fast O(1) check for key existence

  • Prevents errors: Won't try to close non-existent connections

  • Branching logic: Different handling for found vs. not found connections


Step 3: Safe Connection Closure (Success Path)

        manager = connection_pool[connection_key]
        manager.close()
        del connection_pool[connection_key]

The Three-Step Cleanup:

  1. manager = connection_pool[connection_key]: Get the DatabaseManager instance

  2. manager.close(): Properly close the database connection

  3. del connection_pool[connection_key]: Remove from pool to prevent reuse


Why this order matters:

  • Get first: Ensures we have a valid manager object

  • Close second: Properly releases database resources (file locks, sockets, etc.)

  • Delete last: Removes the dead reference from our pool


Step 4: Success Response Construction

        return json.dumps({
            'status': 'success',
            'message': f'Closed connection to {db_path}',
            'database_path': db_path,
            'database_type': db_type
        }, indent=2)

The Confirmation Receipt:

  • Clear success status: Confirms the operation completed

  • Descriptive message: Human-readable confirmation

  • Context details: Database path and type for logging/debugging

  • Consistent format: Matches response structure of other tools


Step 5: Graceful "Not Found" Handling

    else:
        return json.dumps({
            'status': 'info',
            'message': f'No active connection found for {db_path}',
            'database_path': db_path,
            'database_type': db_type
        }, indent=2)

The "Nothing to Clean" Response:

  • Status 'info': Not an error, just informational

  • Helpful message: Explains why nothing was closed

  • Same context: Includes database details for consistency

  • No exception: Gracefully handles the "not found" case


What manager.close() Actually Does

    def close(self):
        """Close database connection"""
        if self.connection:
            self.connection.close()
            self.is_initialized = False
            logger.info("Database connection closed")

The Internal Cleanup Process:

  • Connection check: Only closes if connection exists

  • self.connection.close(): Calls the database driver's close method

  • State reset: Marks manager as uninitialized

  • Logging: Records the closure for debugging


Resource cleanup includes:

  • File locks released (SQLite)

  • Network sockets closed (PostgreSQL)

  • Memory buffers freed

  • Transaction rollbacks (if needed)


The Grand Finale - Server Startup and Configuration

if __name__ == "__main__":
    print("Starting Path-Based Database Query MCP Server...")
    print("Key Features:")
    print("- Path-based database operations (no global state)")
    print("- Connection pooling per database path")
    print("- Automatic sample database detection")
    print("- Support for multiple concurrent databases")
    print("- PostgreSQL support (if psycopg2 installed)")
    print("\nCore Tools:")
    print("- query_database(db_path, query) - Query any database")
    print("- insert_into_database(db_path, table, data) - Insert into any database")
    print("- update_database(db_path, table, data, where) - Update any database")
    print("- delete_from_database(db_path, table, where) - Delete from any database")
    print("- get_database_tables(db_path) - List tables in any database")
    print("- get_table_schema(db_path, table) - Get table structure")
    print("- list_database_connections() - See active connections")
    print("\nUsage: Always specify db_path in each function call")
    print("Sample DB: './data/sample_database.db'")
    print("Your DB: '/path/to/your/database.db'")
    mcp.run()

The Server Launch Sequence! 🚀

When you run this script, it becomes a fully-featured MCP server that other applications (including AI systems) can connect to and use.

The startup messages serve multiple purposes:

  1. User education: Shows exactly what capabilities are available

  2. Quick reference: Lists all tools with their basic syntax

  3. Usage guidance: Explains the path-based approach

  4. Examples: Provides sample database path and your own database path format

mcp.run(): This is where the magic happens! It starts the MCP server and begins listening for tool requests from AI systems or other clients.


Bringing It All Together


This is the full code for server:


This is requirements.txt file which has all the required libraries to run this code.


Output



MCP Set-Up

First, we will save the files in a folder. In our case, it is named database_mcp. We will have our database_mcp.py file, requirements.txt file, and a folder that contains all the database files.

ree

There are some database files stored in the data folder.

ree

Now we will open the database_mcp directory in the terminal, create a virtual environment, and then activate it.

ree

To create a virtual environment:

  • On Windows: 

python -m venv myenv

  • On Linux and macOS: 

python3 -m venv myenv

Now we will install uv using,

pip install uv

We will also upgrade pip using:

python.exe -m pip install --upgrade pip

uv is an extremely fast Python package and project manager. It is recommended in the documentation of MCP, and we might need it while running the app in development mode.

ree

Now we will install mcp using,

 pip install mcp

It will help us create the MCP server.

ree

Now we will install all the required libraries using,

 pip install -r requirements.txt
ree

Now we can run our code in developer mode using:

mcp dev database_mcp.py

It will provide us with a link that we can open in the browser.

ree

After this, we will click on Connect to connect to the server.

ree

We can go to Tools to view all our tools and perform operations from there.

ree

That is for the development mode. Now we will run it on the Claude Desktop. First, open Claude Desktop and go to the Developer Settings by navigating to: Menu → File → Settings → Developer.

ree

Open the claude_desktop_config.json file.

ree


And now add this configuration, making sure to change the paths according to your environment:

{
  "mcpServers": {
    "database-mcp": {
      "command": "E:/workspace/python/mcp/database_mcp/myenv/Scripts/python.exe",
      "args": [
        "E:/workspace/python/mcp/database_mcp/database_mcp.py"
      ],
      "env": {
        "PYTHONPATH": "E:/workspace/python/mcp/database_mcp/myenv/Lib/site-packages"
      }
    }
  }
}

After doing that, close the Claude app and reopen it. Then open the Developer Settings again; the MCP server should be running. If it is not running, terminate the Claude Desktop app from the Task Manager and reopen it.

ree

Now go to Search and Tools to check if the MCP server and its associated tools are available.

ree

In this case, all the tools are available.

ree

We will now perform some queries.

ree
ree

ree

As we can see from the result, our MCP server is working.



Transform Your Projects with Codersarts

  • Enterprise-grade database servers with advanced security and performance optimization

  • Multi-database integration systems supporting SQLite, PostgreSQL, MySQL, and cloud databases

  • AI-powered query optimization and intelligent database management tools

  • Production-ready deployment with monitoring, logging, and automated backup systems

Don't let complex implementations slow down your innovation. Connect with Codersarts today and turn your ideas into reality!


Ready to get started? Visit Codersarts.com or reach out to our team to discuss your next project. The future of intelligent automation is here – let's build it together!


ree

Comments

bottom of page