Building an MCP Server for Databases: Safe CRUD Tools with SQLite and MySQL

1. Introduction: Should an AI Be Allowed to Touch a Database?

Connecting MCP tools to a database enables requests such as:

Show me the incomplete tasks that are overdue.
Mark task 42 as done.
Delete the old test task.

The attractive but unsafe implementation is to translate those instructions into SQL and execute it. The risk goes well beyond classic SQL injection: an incorrect UPDATE predicate, an unbounded SELECT, or prompt injection that induces access to sensitive records can all cause damage.

The safe boundary is not “the model writes SQL.” It is “the model selects a server-defined use case.” The requests above should map to list_tasks, update_task, and prepare_delete_task. SQL statements and database credentials remain inside the MCP server.

This article uses Python 3.12+, MCP Python SDK v1, SQLite, and MySQL Connector/Python. As of July 16, 2026, the latest v1 release on PyPI is 1.28.1, while v2 remains a prerelease. Add an upper bound to prevent an unintended major upgrade.

dependencies = [
    "mcp[cli]>=1.28,<2",
    "mysql-connector-python",
    "pydantic-settings",
]

2. Connecting an MCP Server Directly to a Database

The model produces only a tool name and structured arguments. The MCP server validates that input and executes SQL that is fixed or assembled exclusively from allowlisted fragments.

flowchart LR
    accTitle: MCP database operation boundary
    accDescr: The AI client turns a user's natural-language request into a tool call. The MCP server validates it and executes only approved SQL against SQLite or MySQL. Credentials and SQL remain inside the server.
    U[User] -->|Natural language| C[AI client]
    C -->|tools/call| M[MCP server]
    M --> V[Schema and domain validation]
    V --> R[Repository]
    R -->|Approved SQL| D[(SQLite / MySQL)]
    M -.->|No credentials exposed| C

The secure boundary from MCP to the database

A directly connected MCP server is effectively a new backend service. It becomes responsible for authorization, auditing, error handling, connection management, and compatibility with database migrations.

Direct access is reasonable for an MCP-owned database, a local SQLite application, a read-only analytics database, or a proof of concept with no existing API. If a production service already has a business API, use it by default. Letting MCP update the database directly can bypass the API’s validation, per-user authorization, notifications, and event publication.

3. The Server We Are Building

The sample domain is task management. The tasks table stores current state, and task_events records created, updated, and deleted events.

The DATABASE_BACKEND=sqlite|mysql environment variable selects a repository without changing tool names or schemas. No arbitrary SQL tool is exposed. Each state change and its audit event are committed in one transaction.

Complete code: The SQLite and MySQL implementations, Docker Compose setup, and test suite are available in the GitHub repository. The filenames shown above the remaining code blocks correspond to paths in that repository.

4. Separating Database Code from Tools

Putting SQL directly in tool functions mixes input validation, output formatting, SQL dialects, and transaction management. Introduce a repository boundary instead.

from typing import Protocol

class TaskRepository(Protocol):
    def list_tasks(self, query: "ListTasksInput") -> list["Task"]: ...
    def get_task(self, task_id: int) -> "Task | None": ...
    def create_task(self, data: "CreateTaskInput") -> "Task": ...
    def update_task(self, data: "UpdateTaskInput") -> "Task | None": ...
    def delete_task(self, task_id: int, expected_updated_at: str) -> bool: ...

SQLiteTaskRepository and MySQLTaskRepository implement this contract. Tools can focus on validation and safe result formatting, while placeholder syntax and lastrowid behavior stay inside each repository. The same contract tests can exercise both implementations, and a future ApiTaskRepository can replace them without redesigning the tools.

5. Managing SQLite Connections

For a small SQLite deployment, opening and closing a connection for each tool invocation is a clear design. Do not casually share one global connection across worker threads. By default, Python’s sqlite3 raises ProgrammingError when a connection is used from a thread other than the one that created it.

from contextlib import contextmanager
import sqlite3

@contextmanager
def sqlite_connection(path: str, *, read_only: bool = False):
    database = f"file:{path}?mode=ro" if read_only else path
    connection = sqlite3.connect(database, uri=read_only)
    connection.row_factory = sqlite3.Row
    try:
        yield connection
        if not read_only:
            connection.commit()
    except Exception:
        if not read_only:
            connection.rollback()
        raise
    finally:
        connection.close()

Use mode=ro for read tools. SQLite rejects writes through a connection opened with this URI, giving you a stronger boundary than code organization alone. Also note that using a Connection as a context manager commits or rolls back a transaction but does not close the connection itself; call close() explicitly.

6. MySQL Connection Pools

For MySQL, avoid establishing a new TCP connection for every tool invocation. Use connection pools, and separate reader and writer users as well as their pools.

from mysql.connector.pooling import MySQLConnectionPool

reader_pool = MySQLConnectionPool(
    pool_name="mcp_task_reader",
    pool_size=5,
    host=settings.mysql_host,
    user=settings.mysql_reader_user,
    password=settings.mysql_reader_password,
    database=settings.mysql_database,
)

writer_pool = MySQLConnectionPool(
    pool_name="mcp_task_writer",
    pool_size=5,
    host=settings.mysql_host,
    user=settings.mysql_writer_user,
    password=settings.mysql_writer_password,
    database=settings.mysql_database,
)

Grant mcp_reader only SELECT on the required tables. Grant mcp_writer only the required SELECT, INSERT, UPDATE, and DELETE privileges. Neither account needs administrator or DDL privileges.

connection = writer_pool.get_connection()
try:
    # Repository operation
    connection.commit()
except Exception:
    connection.rollback()
    raise
finally:
    connection.close()

For a pooled Connector/Python connection, close() returns the connection to the pool rather than physically disconnecting it. It must run on every exception path. An exhausted pool raises PoolError, so size the pool against expected concurrency and the server’s connection limit, and monitor exhaustion.

7. Preventing SQL Injection

Never construct value predicates by string interpolation.

sql = f"SELECT * FROM tasks WHERE status = '{status}'"

Bind all values as parameters instead.

# SQLite
cursor.execute(
    "SELECT * FROM tasks WHERE status = ?",
    (status,),
)

# MySQL
cursor.execute(
    "SELECT * FROM tasks WHERE status = %s",
    (status,),
)

Parameterization keeps SQL code separate from data. OWASP identifies prepared statements and parameterized queries as a primary defense and strongly discourages relying on escaping alone.

Table names, column names, operators, and keywords such as ASC and DESC generally cannot be placeholders. Map enums to constants for those fragments.

SORT_COLUMNS = {
    TaskSort.CREATED_AT: "created_at",
    TaskSort.DUE_DATE: "due_date",
}
SORT_ORDERS = {
    SortOrder.ASC: "ASC",
    SortOrder.DESC: "DESC",
}

Do not ask whether user input “looks dangerous.” Select only known-safe values defined by the application.

8. Separating Read and Write Tools

A generic tool such as manage_tasks(operation, values) has a broad schema, is harder for a model to select correctly, and hides destructiveness in logs. Use-case-specific tools allow narrower inputs, privileges, and audit labels.

The MCP specification defines every Tool Annotation property as a hint. It also says clients should not base tool-use decisions on annotations from untrusted servers. destructiveHint is useful for confirmation UI, but it is not an authorization control and cannot replace database privileges.

Build defense in depth from the input schema, application validation, parameterized queries, database permissions, and transaction boundaries.

9. Implementing the CRUD Tools

Input models narrow the values the model can generate.

from datetime import date
from enum import StrEnum
from typing import Annotated
from pydantic import BaseModel, Field, model_validator

class TaskStatus(StrEnum):
    TODO = "todo"
    IN_PROGRESS = "in_progress"
    DONE = "done"

class ListTasksInput(BaseModel):
    status: TaskStatus | None = None
    limit: Annotated[int, Field(ge=1, le=100)] = 20
    offset: Annotated[int, Field(ge=0)] = 0
    sort_by: TaskSort = TaskSort.CREATED_AT
    sort_order: SortOrder = SortOrder.DESC

class CreateTaskInput(BaseModel):
    title: Annotated[str, Field(min_length=1, max_length=200)]
    description: Annotated[str, Field(max_length=5000)] = ""
    status: TaskStatus = TaskStatus.TODO
    due_date: date | None = None

class UpdateTaskInput(BaseModel):
    task_id: Annotated[int, Field(gt=0)]
    title: Annotated[str, Field(min_length=1, max_length=200)] | None = None
    description: Annotated[str, Field(max_length=5000)] | None = None
    status: TaskStatus | None = None
    due_date: date | None = None

    @model_validator(mode="after")
    def require_change(self):
        if not any(
            value is not None
            for value in (self.title, self.description, self.status, self.due_date)
        ):
            raise ValueError("at least one field must be updated")
        return self

list_tasks caps responses at 100 rows and chooses sorting from an allowlist. A missing record in get_task is not a database failure; return a normal result such as {"found": false}. Internal exceptions may be logged, but tool results must not include SQL text, filesystem paths, hostnames, or credentials.

After create_task inserts a row, use lastrowid and fetch the created record on the same connection. update_task accepts an explicit schema rather than an unrestricted dictionary.

@mcp.tool(annotations={"readOnlyHint": False, "destructiveHint": True})
def update_task(input: UpdateTaskInput) -> dict:
    task = repository.update_task(input)
    if task is None:
        return {
            "updated": False,
            "reason": "not_found_or_conflict",
            "task_id": input.task_id,
        }
    return {"updated": True, "task": task.model_dump(mode="json")}

To distinguish conflicts precisely, include expected_updated_at or a version number and use optimistic locking with WHERE id = ? AND updated_at = ?. If rowcount == 0, fetch the task again to classify the outcome as missing or concurrently modified.

10. Transactions and Rollbacks

update_task must update the task and append its audit event in one transaction.

try:
    updated = update_task_row(connection, input)
    if not updated:
        raise TaskConflictError(input.task_id)

    insert_task_event(
        connection,
        task_id=input.task_id,
        event_type="updated",
        event_data=changes_json,
    )
    connection.commit()
except Exception:
    connection.rollback()
    raise

If the event insert fails, the task update is rolled back as well. Do not commit between the two operations. Connector/Python disables autocommit by default, so successful writes require an explicit commit(), while failures require rollback(). Give the SQLite repository the same transaction boundary through its connection context.

Keeping transaction management in the repository makes it harder to forget the audit write. Avoid calling email services or external APIs inside a database transaction: doing so extends lock time and can leave the external action completed after the database rolls back. Use an outbox pattern when needed.

11. Requiring Confirmation for Deletes

A confirm=True argument is not confirmation because the model can generate the Boolean itself. Bind the exact record, its version, the operation, and an expiry time into a signed token created immediately before deletion.

sequenceDiagram
    accTitle: Delete confirmation with a signed token
    accDescr: The AI client calls the preparation tool, and the server returns target details plus a short-lived signed token. After user approval, the deletion tool validates the signature, expiry, target, record timestamp, and unused nonce before deleting the task and writing its audit event in one transaction.
    actor U as User
    participant C as AI client
    participant M as MCP server
    participant D as Database
    U->>C: Delete task 42
    C->>M: prepare_delete_task(42)
    M->>D: Read task and updated_at
    D-->>M: Target record
    M-->>C: Target details and signed token
    C-->>U: Delete this task?
    U->>C: Delete it
    C->>M: confirm_delete_task(42, token)
    M->>M: Validate signature, expiry, ID, version, and nonce
    M->>D: DELETE and audit INSERT
    D-->>M: COMMIT
    M-->>C: Deletion completed

Two-step deletion confirmation sequence

The payload should contain operation=delete_task, task_id, updated_at, an expiry field such as exp, and a random nonce, then be signed with a server-held HMAC key. confirm_delete_task verifies that:

• the signature is valid and the token has not expired;
• the argument and token refer to the same task;
• the current updated_at matches the version in the token;
• the operation is delete_task; and
• the nonce has not already been consumed.

Reject modified, expired, cross-task, stale, and replayed tokens. Consume the nonce, delete the task, and insert the deleted event in the same transaction. Do not log the raw token, and load the signing key from environment-backed or managed secret storage.

12. Why You Should Not Expose Arbitrary SQL

The following tool merely exposes a database driver through MCP.

@mcp.tool()
def execute_sql(sql: str) -> list[dict]:
    ...

Its schema cannot constrain the operation. In addition to DROP or an unqualified DELETE, it permits large result sets, access to sensitive columns and system tables, expensive joins, and locking queries. Its name does not reveal the risk, and an audit log can classify the action only as “executed SQL.”

A “SELECT-only” variant is still a poor default. Reliably proving that arbitrary SQL is read-only is nontrivial, and read operations can still leak information or exhaust resources. Prefer tools such as get_overdue_tasks, count_tasks_by_status, and get_task_activity, where the server controls columns, predicates, row limits, and execution time.

13. Testing the Security Properties

Run the same repository contract tests against SQLite and MySQL. At minimum, verify that the implementation:

• handles existing and missing tasks correctly;
• rejects invalid statuses and limit=1000 at the schema boundary;
• records create, update, and delete events;
• converts database failures into safe tool errors;
• rejects writes through a read-only connection.

For SQL injection tests, submit ' OR 1=1 -- as a title or search value. Confirm that it is stored or compared as data and never changes the SQL structure. The goal is not to reject suspicious punctuation; it is to handle legitimate strings safely.

For the transaction test, force the event insert to fail after updating the task, then verify that the task returned to its original state. Delete tests should cover modified, expired, cross-task, and stale tokens, plus replay of the same token.

Start a real MySQL instance with Docker Compose and run the same contract suite used for SQLite. Placeholder syntax, temporal types, rowcount, and transaction behavior are easy to miss with mocks alone.

14. Calling the Tools from an AI Client

Use MCP Inspector or a compatible client to verify that natural language produces the expected tool calls.

User: Show the five incomplete tasks with the nearest due dates
AI: list_tasks(status="todo", limit=5, sort_by="due_date", sort_order="asc")

User: Mark task 12 as done
AI: update_task(task_id=12, status="done")

User: Delete task 42
AI: prepare_delete_task(task_id=42)
AI: Delete “Old test task”?
User: Delete it
AI: confirm_delete_task(task_id=42, confirmation_token="...")

Correct tool selection does not mean the caller is authorized. For a remote MCP server, verify on the server that the authenticated principal may access the task. Never trust a user_id supplied as a tool argument; derive the principal from authenticated request context.

15. Production Checklist and Conclusion

• No arbitrary SQL tool is exposed.
• User values are never concatenated into SQL.
• Types, lengths, row counts, and sort fields are constrained.
• Authorization is checked against the authenticated principal.
• Reader and writer connections use separate database privileges.
• Application accounts have no administrator or DDL privileges.
• Connections and cursors are closed on every exception path.
• Related writes share one transaction.
• Tool errors do not reveal SQL or credentials.
• Destructive confirmation is bound to a target, version, and expiry.
• Tool calls and audit events are traceable.
• Rate limits, timeouts, and result limits are configured.
• Existing business APIs are preferred for production services.

An MCP tool is not a thin database wrapper. It is a boundary around permissions and use cases. Instead of giving an AI SQL, give it narrow capabilities such as get_task and update_task, then layer schemas, domain validation, parameterization, least privilege, transactions, auditing, and human confirmation.

That boundary remains the same when moving from SQLite to MySQL. Connection management and SQL dialects should change; the scope of authority granted to the AI should not.