Project 15: Your First MCP Server - SQLite Database Interface

Project Overview

Attribute	Value
File	P15-sqlite-mcp-server.md
Main Programming Language	Python
Alternative Programming Languages	TypeScript/Node, Go, Rust
Coolness Level	Level 3: Genuinely Clever
Business Potential	3. The “Service & Support” Model
Difficulty	Level 2: Intermediate
Knowledge Area	MCP / Databases / Protocol Implementation
Software or Tool	MCP SDK, SQLite
Main Book	“Designing Data-Intensive Applications” by Martin Kleppmann
Time Estimate	1 week
Prerequisites	Python basics, SQL basics, JSON-RPC understanding

What You Will Build

A Model Context Protocol server that exposes SQLite database operations as tools Claude can invoke: query, insert, update, delete, and schema inspection. Includes proper error handling and query sanitization.

This is your “Hello World” for MCP servers. You will understand the protocol, tool definitions, stdio transport, and how Claude invokes external services.

Real World Outcome

You: What tables are in my database?

Claude: [Invokes mcp__sqlite__list_tables]

Your database has the following tables:
- users (5 columns, 1,234 rows)
- orders (8 columns, 5,678 rows)
- products (6 columns, 342 rows)

You: Show me the top 5 customers by order count

Claude: [Invokes mcp__sqlite__query]

Here are your top customers:

| Customer | Orders | Total Spent |
|----------|--------|-------------|
| Alice    | 45     | $3,450.00   |
| Bob      | 38     | $2,890.00   |
| Carol    | 32     | $2,100.00   |
...

The Core Question You Are Answering

“How do I extend Claude’s capabilities by creating a service that exposes tools via the Model Context Protocol?”

MCP is Claude’s extension mechanism. By building an MCP server, you give Claude the ability to interact with ANY external system: databases, APIs, hardware, anything. Understanding this protocol opens the door to building powerful AI integrations.

MCP Architecture Overview

+------------------------------------------------------------------+
|                        CLAUDE CODE                                 |
|                                                                    |
|   +------------------------------------------------------------+   |
|   |                    MCP CLIENT                               |   |
|   |                                                             |   |
|   |   Reads .mcp.json -> Spawns MCP servers -> Invokes tools   |   |
|   +------------------------------------------------------------+   |
|                               |                                    |
|                               | stdio (JSON-RPC 2.0)               |
|                               |                                    |
+-------------------------------v------------------------------------+
                                |
+-------------------------------v------------------------------------+
|                       YOUR MCP SERVER                              |
|                                                                    |
|   +------------+    +------------+    +------------------+         |
|   | list_tools |    | call_tool  |    | list_resources   |         |
|   +------------+    +------------+    +------------------+         |
|                           |                                        |
|                           v                                        |
|                  +------------------+                              |
|                  |     SQLite DB    |                              |
|                  +------------------+                              |
|                                                                    |
+--------------------------------------------------------------------+

stdio Transport Deep Dive

+-------------------+                    +-------------------+
|    CLAUDE CODE    |                    |  YOUR MCP SERVER  |
|                   |                    |                   |
|  stdin  <---------+--------------------+----- stdout      |
|  stdout +---------+------------------->+------ stdin      |
|                   |                    |                   |
+-------------------+                    +-------------------+
        |                                         |
        |    JSON-RPC 2.0 Messages                |
        |    {"jsonrpc":"2.0","method":"...",     |
        |     "params":{...},"id":1}              |
        |                                         |
        +-- Request --->  (your server reads)     |
        <-- Response ---  (your server writes)    |

The stdio transport is:

Synchronous per message: One request, one response
Line-delimited: Each JSON-RPC message is one line
Bidirectional: Claude writes to your stdin, reads from your stdout

Concepts You Must Understand First

Stop and research these before coding:

1. MCP Protocol Basics

What is JSON-RPC and how does version 2.0 work?
How do tools differ from resources in MCP?
What is the lifecycle of an MCP request (initialization, tool discovery, invocation)?
Reference: spec.modelcontextprotocol.io

2. Transport Types

Reference: Claude Code Docs: MCP section

3. Tool Definition Schema

How do you define input parameters using JSON Schema?
How do you specify output format?
What are the error handling conventions?
Reference: MCP SDK documentation

4. SQL Injection Prevention

What is SQL injection and why is it dangerous?
How do parameterized queries prevent injection?
What is the principle of least privilege for database access?
Reference: OWASP SQL Injection Prevention Cheat Sheet

Questions to Guide Your Design

Before implementing, think through these:

1. What Tools Should You Expose?

Tool	Purpose	Risk Level
`list_tables`	Get database schema	Low (read-only)
`describe_table`	Get column details	Low (read-only)
`query`	Run SELECT queries	Medium (data exposure)
`execute`	Run INSERT/UPDATE/DELETE	High (data modification)

2. Security Considerations

Should you allow arbitrary SQL or only parameterized queries?
How do you prevent SQL injection in a natural language interface?
Should write operations require explicit confirmation?
Should you whitelist allowed tables/columns?

3. Configuration Strategy

How do you specify which database to connect to?
Environment variables vs command-line arguments?
Should the database path be hardcoded or configurable?

Thinking Exercise

Design the Tool Schema Before Implementing

Define your tools with JSON Schema:

{
  "tools": [
    {
      "name": "list_tables",
      "description": "List all tables in the database with row counts",
      "inputSchema": {
        "type": "object",
        "properties": {},
        "required": []
      }
    },
    {
      "name": "describe_table",
      "description": "Get column names and types for a specific table",
      "inputSchema": {
        "type": "object",
        "properties": {
          "table_name": {
            "type": "string",
            "description": "Name of the table to describe"
          }
        },
        "required": ["table_name"]
      }
    },
    {
      "name": "query",
      "description": "Execute a read-only SQL SELECT query",
      "inputSchema": {
        "type": "object",
        "properties": {
          "sql": {
            "type": "string",
            "description": "The SQL SELECT query to execute"
          },
          "limit": {
            "type": "integer",
            "description": "Maximum rows to return (default: 100)",
            "default": 100
          }
        },
        "required": ["sql"]
      }
    }
  ]
}

Questions to consider:

Should query accept parameters for prepared statements?
How do you return results: JSON array, formatted table, or CSV?
What errors should you surface to Claude vs handle silently?
Should you add a max_limit to prevent memory exhaustion?

The Interview Questions They Will Ask

“How would you extend an AI assistant to interact with a database?”
- Discuss the MCP architecture, tool definitions, and security considerations.
“What is the Model Context Protocol and how does it work?”
- Explain JSON-RPC, stdio transport, tool discovery, and invocation flow.
“How do you prevent SQL injection in a natural language interface?”
- Discuss parameterized queries, input validation, query whitelisting, and least privilege.
“What is the difference between MCP tools and resources?”
- Tools perform actions (side effects), resources provide data (read-only).
“How would you handle authentication for an MCP server?”
- Discuss environment variables, token passing, and credential management.

Hints in Layers

Hint 1: Use the MCP SDK

Do not implement JSON-RPC from scratch. Install the mcp package:

pip install mcp

The SDK handles protocol details so you can focus on your tool logic.

Hint 2: Start with list_tables

Get the simplest tool working first. Return table names as a list:

@server.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "list_tables":
        cursor = conn.execute(
            "SELECT name FROM sqlite_master WHERE type='table'"
        )
        return [TextContent(type="text", text="\n".join(row[0] for row in cursor))]

Hint 3: Configure in .mcp.json

Add your server to Claude’s configuration:

{
  "mcpServers": {
    "sqlite": {
      "type": "stdio",
      "command": "python",
      "args": ["/path/to/server.py", "--db", "mydata.db"]
    }
  }
}

Hint 4: Test with Claude

Ask Claude: “What MCP tools are available?” to verify your server is connected. If it lists your tools, the integration works.

Books That Will Help

Topic	Book	Chapter	Why It Helps
SQL & Databases	“Designing Data-Intensive Applications” by Martin Kleppmann	Ch. 2-3	Understand data models and query languages
Protocol Design	“Building Microservices” by Sam Newman	Ch. 4	Learn service communication patterns
Python Async	“Fluent Python” by Luciano Ramalho	Ch. 21	Master async/await for MCP servers
Security	OWASP Cheat Sheets	SQL Injection	Prevent common vulnerabilities

Implementation Skeleton

#!/usr/bin/env python3
"""SQLite MCP Server - Expose database operations to Claude."""

import argparse
import asyncio
import sqlite3
from pathlib import Path

from mcp.server import Server
from mcp.types import Tool, TextContent
from mcp.server.stdio import stdio_server

# Global database connection (set on startup)
DB_PATH: Path = None
server = Server("sqlite-server")


@server.list_tools()
async def list_tools():
    """Define available tools."""
    return [
        Tool(
            name="list_tables",
            description="List all tables in the database with row counts",
            inputSchema={"type": "object", "properties": {}}
        ),
        Tool(
            name="describe_table",
            description="Get column names, types, and constraints for a table",
            inputSchema={
                "type": "object",
                "properties": {
                    "table_name": {
                        "type": "string",
                        "description": "Name of the table to describe"
                    }
                },
                "required": ["table_name"]
            }
        ),
        Tool(
            name="query",
            description="Execute a read-only SQL SELECT query",
            inputSchema={
                "type": "object",
                "properties": {
                    "sql": {
                        "type": "string",
                        "description": "The SQL SELECT query to execute"
                    }
                },
                "required": ["sql"]
            }
        )
    ]


@server.call_tool()
async def call_tool(name: str, arguments: dict):
    """Handle tool invocations."""
    conn = sqlite3.connect(DB_PATH)

    try:
        if name == "list_tables":
            cursor = conn.execute("""
                SELECT name FROM sqlite_master
                WHERE type='table' AND name NOT LIKE 'sqlite_%'
            """)
            tables = []
            for (table_name,) in cursor.fetchall():
                count = conn.execute(f"SELECT COUNT(*) FROM {table_name}").fetchone()[0]
                tables.append(f"- {table_name} ({count} rows)")
            return [TextContent(type="text", text="\n".join(tables))]

        elif name == "describe_table":
            table_name = arguments["table_name"]
            # Validate table name to prevent injection
            valid_tables = [row[0] for row in conn.execute(
                "SELECT name FROM sqlite_master WHERE type='table'"
            ).fetchall()]
            if table_name not in valid_tables:
                return [TextContent(type="text", text=f"Error: Table '{table_name}' not found")]

            cursor = conn.execute(f"PRAGMA table_info({table_name})")
            columns = []
            for row in cursor.fetchall():
                columns.append(f"- {row[1]} ({row[2]})" + (" PRIMARY KEY" if row[5] else ""))
            return [TextContent(type="text", text=f"Table: {table_name}\n" + "\n".join(columns))]

        elif name == "query":
            sql = arguments["sql"].strip()

            # Security: Only allow SELECT queries
            if not sql.upper().startswith("SELECT"):
                return [TextContent(
                    type="text",
                    text="Error: Only SELECT queries are allowed. Use other tools for modifications."
                )]

            cursor = conn.execute(sql)
            columns = [desc[0] for desc in cursor.description] if cursor.description else []
            rows = cursor.fetchall()

            # Format as markdown table
            if not rows:
                return [TextContent(type="text", text="Query returned no results.")]

            result = "| " + " | ".join(columns) + " |\n"
            result += "|" + "|".join(["---"] * len(columns)) + "|\n"
            for row in rows[:100]:  # Limit output
                result += "| " + " | ".join(str(v) for v in row) + " |\n"

            if len(rows) > 100:
                result += f"\n... and {len(rows) - 100} more rows"

            return [TextContent(type="text", text=result)]

        else:
            return [TextContent(type="text", text=f"Unknown tool: {name}")]

    finally:
        conn.close()


async def main():
    global DB_PATH

    parser = argparse.ArgumentParser(description="SQLite MCP Server")
    parser.add_argument("--db", required=True, help="Path to SQLite database")
    args = parser.parse_args()

    DB_PATH = Path(args.db)
    if not DB_PATH.exists():
        print(f"Error: Database not found: {DB_PATH}", file=sys.stderr)
        sys.exit(1)

    async with stdio_server() as (read_stream, write_stream):
        await server.run(read_stream, write_stream, server.create_initialization_options())


if __name__ == "__main__":
    import sys
    asyncio.run(main())

Learning Milestones

Track your progress through these checkpoints:

Milestone	What It Proves	Verification
Server starts and responds	You understand MCP basics	No errors on startup
Claude can list tables	Tool invocation works	Ask “What tables exist?”
Queries return formatted results	You built a useful MCP server	Ask “Show me all users”
Error handling is robust	Production-ready implementation	Invalid SQL returns helpful error

Core Challenges Mapped to Concepts

Challenge	Concept	Book Reference
Implementing the MCP protocol	JSON-RPC and tool schemas	MCP Specification
Exposing database operations	Tool definition design	“Building Microservices” Ch. 4
Handling SQL injection	Parameterized queries	OWASP Cheat Sheets
Configuring Claude to use your server	.mcp.json setup	Claude Code Docs

Extension Ideas

Once the basic server works, consider these enhancements:

Add write operations with confirmation prompts
Implement query caching for repeated queries
Add query explain to show execution plans
Support multiple databases via resource URIs
Add transaction support for multi-step operations

Common Pitfalls

Forgetting to close database connections - Use try/finally or context managers
Not validating table names - SQL injection via table names is possible
Returning too much data - Always limit results to prevent context overflow
Blocking the event loop - Use asyncio.to_thread() for SQLite operations in production
Hardcoding database paths - Make configuration flexible via arguments or environment

Success Criteria

You have completed this project when:

Your MCP server starts without errors
Claude can discover your tools with “What MCP tools are available?”
list_tables returns all tables with row counts
describe_table shows column information
query executes SELECT statements and returns formatted results
Non-SELECT queries are rejected with a helpful error message
Invalid table names are handled gracefully
The server can be configured via command-line arguments