MySQL MCP Toolkit

I started this as a 3-tool prototype (read_tasks, insert_task, update_task) hardcoded to a single table, with a password sitting right there in the source. It worked, but it was the kind of thing you'd never want an AI model actually calling against a real database. This is the rewrite — a general-purpose MySQL toolkit exposed over MCP that works on any table, validates everything before it touches the database, and doesn't let a model nuke a table by accident.

MCP

MCP (Model Context Protocol) is just a standard way for an AI model to call functions you've written, without you having to hand-roll a different integration for every provider. You run an MCP server that says "here are the tools I have, here's what each one expects." An MCP client — usually wired into an AI model — connects to that server, reads the tool list, and lets the model decide which tool to call and with what arguments.

The model never sees your database or writes any SQL itself. It only ever sees tool names and their input schemas. Whatever it asks for still has to pass through your own validation code before anything actually happens — that boundary is basically the whole point of this project.

Roughly, a request looks like this:

sequenceDiagram
    participant U as User
    participant AI as AI Model (Gemini, in this repo)
    participant C as MCP Client
    participant S as MCP Server (db_mcp_server.py)
    participant DB as MySQL

    U->>AI: "Show me out-of-stock products"
    AI->>C: call query_table_tool(table="products", filters={...})
    C->>S: tool request
    S->>DB: validated, parameterized SQL
    DB-->>S: rows
    S-->>C: JSON result
    C-->>AI: tool result
    AI-->>U: answer in plain English

What changed from the prototype

Before	After
New MySQL connection per call	Connection pool (db_pool.py)
Hardcoded password in source	Loaded from .env (config.py)
Hardcoded Gemini API key in source	Loaded from .env
Only worked on the tasks table	Works on any table via live schema discovery
Raw SQL built from whatever input came in	Table/column names checked against a real schema whitelist first
update_task just ran, instantly	update_row / delete_row need a filter and a second confirm call
No logging at all	Every call gets written to audit_log
Unbounded SELECT *	Always paginated, always capped by MAX_QUERY_LIMIT

How the pieces fit together

flowchart TB
    subgraph Client["🖥️ Client Side"]
        GC["gemini_mcp_client.py<br/>example agent loop"]
    end

    subgraph Server["⚙️ MCP Server Process"]
        SRV["db_mcp_server.py<br/>registers @mcp.tool() functions"]
        TOOLS["db_tools.py<br/>discover / query / insert / update / delete / restore"]
        CACHE["schema_cache.py<br/>live INFORMATION_SCHEMA whitelist"]
        AUDIT["audit.py<br/>writes to audit_log"]
        CFG["config.py<br/>reads .env"]
        POOL["db_pool.py<br/>MySQL connection pool"]
    end

    subgraph DB["🗄️ MySQL"]
        TBL["your tables"]
        LOG["audit_log"]
    end

    GC -- "stdio (MCP)" --> SRV
    SRV --> TOOLS
    TOOLS --> CACHE
    TOOLS --> AUDIT
    TOOLS --> POOL
    CACHE --> POOL
    AUDIT --> POOL
    POOL --> CFG
    POOL --> TBL
    AUDIT --> LOG

    classDef clientNode fill:#E3F2FD,stroke:#1565C0,stroke-width:1.5px,color:#0D47A1
    classDef serverNode fill:#FFF3E0,stroke:#EF6C00,stroke-width:1.5px,color:#E65100
    classDef securityNode fill:#F3E5F5,stroke:#6A1B9A,stroke-width:1.5px,color:#4A148C
    classDef dbNode fill:#E8F5E9,stroke:#2E7D32,stroke-width:1.5px,color:#1B5E20

    class GC clientNode
    class SRV,TOOLS,CFG,POOL serverNode
    class CACHE,AUDIT securityNode
    class TBL,LOG dbNode

    style Client fill:#F5FAFF,stroke:#90CAF9,stroke-width:1px
    style Server fill:#FFFDF7,stroke:#FFCC80,stroke-width:1px
    style DB fill:#F5FFF6,stroke:#A5D6A7,stroke-width:1px

Blue is the client, orange is the core server logic, purple is the two files doing the actual safety work (schema_cache.py and audit.py), and green is the database layer. A quick tour of what each file is actually doing:

• config.py — reads everything (DB_HOST, DB_PORT, DB_USER, DB_PASSWORD, DB_NAME, DB_POOL_SIZE, MAX_QUERY_LIMIT, GEMINI_API_KEY) from .env. It also prints a startup fingerprint to stderr on launch, so if you're ever confused about which database a running process is actually pointed at, check the logs — it's right there.
• db_pool.py — one shared MySQLConnectionPool instead of opening a new connection per call. Everything else just asks this for a connection.
• schema_cache.py — this is the file doing the actual security work. Table and column names can't be parameterized the way values can (you can't do WHERE %s = %s for a column name), so instead this reads the real schema from INFORMATION_SCHEMA.COLUMNS on a 60-second cache and every tool checks names against it before they ever get near a query string.
• db_tools.py — the actual logic: discover_tables, describe_table, query_table, insert_row, update_row, delete_row, restore_row. Everything here validates against schema_cache, uses %s placeholders for values, and logs its own result.
• db_mcp_server.py — wraps each function above as an @mcp.tool(), catches errors into a consistent {"status": "error", ...} shape, and runs the server over stdio.
• audit.py — creates and writes to audit_log. Every call, success or failure, params, and a result summary.
• gemini_mcp_client.py — an example client, not something the server needs to run. It shows an actual agent loop: Gemini calls a tool, gets a result, decides whether to call another one or just answer. It also handles a real annoyance — Gemini's function-calling schema doesn't understand additionalProperties or anyOf, both of which MCP generates automatically from Python type hints, so clean_schema() strips those out before anything gets sent to Gemini. This is Gemini-specific; if you want to hook this server up to a different model, you'd write your own client, but the server side doesn't change at all.
• schema.sql — a sample products table plus audit_log, with a few rows seeded in so there's something to query right away.

Following one call through, start to finish

Take update_row_tool(table="products", filters={"id": 3}, values={"stock_quantity": 0}):

flowchart LR
    A["AI calls update_row_tool"] --> B{"Table exists?"}
    B -- no --> E1["ValidationError"]
    B -- yes --> C{"filters given?"}
    C -- no --> E2["refuse — would hit the whole table"]
    C -- yes --> D{"columns valid?"}
    D -- no --> E3["ValidationError"]
    D -- yes --> F["count matching rows"]
    F --> G{"confirm=True?"}
    G -- no --> H["return preview + row count"]
    G -- yes --> I["run parameterized UPDATE"]
    I --> J["commit, write to audit_log"]
    J --> K["return rows_updated"]

delete_row and restore_row follow the same preview-then-confirm shape. Deletes are soft by default when the table has an is_deleted column (pass hard=True to actually remove the row), and restore_row_tool will throw a ValidationError if you try it on a table that doesn't have soft-delete support in the first place.

Safety, in plain terms

• No SQL injection — names go through the schema_cache whitelist, values always go through %s placeholders. No string concatenation, anywhere.
• No accidental mass writes — update_row_tool and delete_row_tool won't run without a filter, and both need a second call with confirm=True before anything actually happens.
• No runaway queries — query_table_tool is always paginated and capped by MAX_QUERY_LIMIT, no matter what page_size gets passed in.
• Everything's logged — every call, whether it succeeded or blew up, goes into audit_log with the tool name, params, and a summary.
• Soft delete by default — if the table supports it, deletes flag rows instead of removing them, and can be undone.

Setup

1. Create a .env file in the project root with your own values:

   DB_HOST=localhost
   DB_PORT=3306
   DB_USER=root
   DB_PASSWORD=your_password
   DB_NAME=products_db
   DB_POOL_SIZE=5
   MAX_QUERY_LIMIT=100
   GEMINI_API_KEY=your_gemini_key
   

2. Install dependencies:

   pip install -r requirements.txt
   

3. Load the sample schema — note it creates data in a products_db database to match the default DB_NAME above (change both if you're pointing at your own DB):

   mysql -u root -p -e "CREATE DATABASE IF NOT EXISTS products_db"
   mysql -u root -p products_db < schema.sql
   

4. Run the server:

   python db_mcp_server.py
   

5. Optionally, try the example Gemini client against it:

   python gemini_mcp_client.py
   

Tools this exposes

Tool	What it does	Guardrail
list_tables_tool	lists every table	read-only
describe_table_tool	columns + soft-delete support	read-only
query_table_tool	paginated SELECT with filters/columns	capped by MAX_QUERY_LIMIT, hides soft-deleted rows by default
insert_row_tool	inserts a row	columns validated
update_row_tool	updates matching rows	needs a filter + confirm=True
delete_row_tool	soft or hard deletes matching rows	needs a filter + confirm=True
restore_row_tool	undoes a soft delete	needs a filter + confirm=True