Model Context Protocol Server Architecture with Python and SQLite

DEEP DIVE SERIES · #03

Giving the agent hands: wiring local MCP servers with Python and SQLite

6 min read

Post one was the lab. Post two was the map. Now we give the agent hands.

An LLM trapped in a chat box can generate persuasive text, but it can’t inspect a database, verify a record, or touch a filesystem. To do real work, an agent needs tools.

Until recently, every framework had its own incompatible way of defining tools — LangChain tools didn’t talk to CrewAI tools, and neither talked to Google ADK or Claude Desktop. Then Anthropic open-sourced the Model Context Protocol (MCP), and the industry finally got a standardized, client-server contract for AI tooling.

In this build, we’re taking our local lab from Post #1 and wiring up a real, local MCP server from scratch. No hosted intermediaries, no third-party subscription APIs — just Python, stdio transport, and an actual SQLite database.

Here’s the build, mistakes included.

Why MCP matters (and why bespoke tool APIs are dead)

Before MCP, writing a tool meant binding your Python function directly to one model provider’s JSON schema format. If you wanted that same tool in another agent runtime, you rewrote the wrapper.

MCP flips the architecture into a clean client-server model:

  • MCP Host / Client: The agent runtime (your local script or agent framework) that needs to invoke actions.
  • MCP Server: A lightweight, standalone process that exposes three primitives:
    1. Tools: Functions the model can decide to execute (e.g. query_database).
    2. Resources: Data the model can read like files or schemas (e.g. schema://main).
    3. Prompts: Reusable prompt templates with embedded context.
  • Transport: Standard input/output (stdio) for local process isolation, or Server-Sent Events (SSE) for remote services.
Local Model Context Protocol Server with SQLite and Stdio Transport

The beauty of stdio transport is isolation: the MCP server runs as a separate process. If your tool crashes, leaks memory, or encounters a segmentation fault, your agent’s reasoning loop stays alive.

Setting up the MCP server in Python

Let’s build a dedicated SQLite MCP server using the official Python mcp SDK.

Install the required dependencies inside your virtual environment:

uv add "mcp[cli]" aiosqlite

Now create server.py:

import asyncio
import json
import sqlite3
from typing import Any
from mcp.server.fastmcp import FastMCP

# Initialize a FastMCP server named "sqlite-inspector"
mcp = FastMCP("sqlite-inspector")

DB_PATH = "analytics.db"

def init_demo_db():
    conn = sqlite3.connect(DB_PATH)
    cur = conn.cursor()
    cur.execute("""
        CREATE TABLE IF NOT EXISTS agent_runs (
            run_id TEXT PRIMARY KEY,
            agent_name TEXT,
            tokens_used INTEGER,
            latency_ms REAL,
            status TEXT
        )
    """)
    cur.execute("DELETE FROM agent_runs")
    cur.executemany("""
        INSERT INTO agent_runs VALUES (?, ?, ?, ?, ?)
    """, [
        ("run-101", "researcher", 1420, 840.5, "completed"),
        ("run-102", "writer", 3890, 2150.2, "completed"),
        ("run-103", "coder", 5210, 4890.0, "failed_timeout"),
    ])
    conn.commit()
    conn.close()

init_demo_db()

@mcp.tool()
def inspect_schema() -> str:
    """Returns the SQL schema for all tables in the database."""
    conn = sqlite3.connect(DB_PATH)
    cur = conn.cursor()
    cur.execute("SELECT sql FROM sqlite_master WHERE type='table';")
    tables = [row[0] for row in cur.fetchall() if row[0]]
    conn.close()
    return "\n\n".join(tables)

@mcp.tool()
def read_query(sql_query: str) -> str:
    """Executes a read-only SQL query against the database and returns JSON rows."""
    cleaned = sql_query.strip().lower()
    if not cleaned.startswith("select") and not cleaned.startswith("pragma"):
        return json.dumps({"error": "Only SELECT or PRAGMA statements are permitted."})
    
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    cur = conn.cursor()
    try:
        cur.execute(sql_query)
        rows = [dict(row) for row in cur.fetchall()]
        return json.dumps(rows, indent=2)
    except Exception as e:
        return json.dumps({"error": str(e)})
    finally:
        conn.close()

if __name__ == "__main__":
    mcp.run(transport="stdio")

FastMCP gives us type-safe decorators that automatically generate JSON Schema definitions for inspect_schema and read_query without writing a single line of boilerplate schema mapping.

Testing with the MCP Inspector

Never connect a server to an agent without testing the protocol manually first. Anthropic provides an interactive UI inspector:

npx @modelcontextprotocol/inspector uv run server.py

This launches a local dashboard where you can click List Tools, view the auto-generated parameter schemas, and test calling read_query directly with SELECT * FROM agent_runs. If it works in the inspector, it will work with your model.

Connecting the agent client

Now let’s wire our local agent script to consume the server over stdio:

import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def run_agent():
    server_params = StdioServerParameters(
        command="uv",
        args=["run", "server.py"],
        env=None
    )

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # Initialize the handshake
            await session.initialize()

            # 1. Discover available tools
            tools = await session.list_tools()
            print("Connected tools:")
            for tool in tools.tools:
                print(f" - {tool.name}: {tool.description}")

            # 2. Invoke schema inspection
            schema_res = await session.call_tool("inspect_schema")
            print(f"\nDiscovered Schema:\n{schema_res.content[0].text}")

            # 3. Model decides to run an analytical query
            query = "SELECT agent_name, AVG(tokens_used) as avg_tokens FROM agent_runs GROUP BY agent_name"
            result = await session.call_tool("read_query", arguments={"sql_query": query})
            print(f"\nQuery Output:\n{result.content[0].text}")

if __name__ == "__main__":
    asyncio.run(run_agent())

Run it:

uv run client.py

Output:

Connected tools:
 - inspect_schema: Returns the SQL schema for all tables in the database.
 - read_query: Executes a read-only SQL query against the database and returns JSON rows.

Discovered Schema:
CREATE TABLE agent_runs (
    run_id TEXT PRIMARY KEY,
    agent_name TEXT,
    tokens_used INTEGER,
    latency_ms REAL,
    status TEXT
)

Query Output:
[
  {"agent_name": "coder", "avg_tokens": 5210.0},
  {"agent_name": "researcher", "avg_tokens": 1420.0},
  {"agent_name": "writer", "avg_tokens": 3890.0}
]

What broke (The Junky debugging reality)

Nothing in agent engineering works cleanly on the first try. Here are the three traps that caught me:

1. stdout pollution breaks stdio JSON-RPC

If your tool or database initialization contains a stray print("Connecting to DB..."), the entire MCP connection instantly crashes with a parse error.

In stdio mode, standard output belongs exclusively to the JSON-RPC protocol messages. If you need logging inside an MCP server, always log to stderr or a file:

import sys
# CORRECT:
print("Log message", file=sys.stderr)
# WRONG (Crashes client):
print("Log message")

2. The Tool-Selection Explosion Trap

It’s tempting to expose 25 tools to an agent all at once. In practice, when an agent has more than 6–8 tools in its context window:

  • Parameter hallucination rates skyrocket.
  • The model picks slightly wrong tools with similar descriptions.
  • Token costs increase on every single conversational turn.

Engineering fix: Scope tools dynamically. Give the agent an inspection tool first; only expose write/execute tools once the agent has formulated an explicit, validated plan.

3. Read-only query guardrails

LLMs generating SQL queries will eventually attempt DROP TABLE or DELETE when debugging a faulty result. Enforcing read-only constraints in the tool code (or running against a read-only SQLite connection file:analytics.db?mode=ro) is non-negotiable before giving any agent access to databases.

What’s next

We’ve given our local agent hands. It can connect to isolated subprocesses over standard protocols and query structured databases without vendor lock-in.

In Series #4, we put this to the ultimate test: The Build Shootout. We’ll take this exact agent task and build it side-by-side in LangGraph vs Google ADK / AWS Bedrock — measuring boilerplate, debugging headaches, token spend, and production trade-offs.

Build agents. Ship impact. Stay an AgentJunky.

🏛️ Tool Routing & Sandboxing PillarFoundational Knowledge

Engineering Reliable AI Agents: Tool Routing, Permissions and Evaluation

Ground local tool servers in formal security models, scoped access rules, and automated evaluation harnesses.

Newsletter Edition #1

Subscribe to The AgentJunky AI Brief

Get practical thinking on agentic AI, enterprise architecture, RAG, and production governance delivered straight to your LinkedIn feed.

Subscribe on LinkedIn ↗

Discuss an AI Opportunity

Need an architectural review, help transitioning an agent prototype to production, or designing governed tool gateways? Let's connect.