MCP SQLite RBAC Demo

A production-quality demonstration of a Model Context Protocol (MCP) server that securely exposes a SQLite database to AI agents using Role-Based Access Control (RBAC).

Overview

This project teaches developers how to build enterprise MCP servers with:

• Role-Based Access Control (RBAC) - Three roles (admin, manager, viewer) with granular permissions
• Authentication & Authorization - Login/logout with permission enforcement
• Clean Architecture - Tools → Services → Repositories → Database
• Audit Logging - Track all write operations for compliance
• Type Safety - Full type hints with Pydantic schemas
• Input Validation - Comprehensive validation at all layers

Architecture

┌─────────────────────────────────────────┐
│         AI Agent / Claude Client        │
└────────────────┬────────────────────────┘
                 │
                 │ MCP Protocol (JSON-RPC)
                 ▼
┌─────────────────────────────────────────┐
│      MCP Server (FastMCP)               │
│  • auth_tools.py                        │
│  • customer_tools.py                    │
│  • order_tools.py                       │
│  • user_tools.py                        │
└────────────────┬────────────────────────┘
                 │
                 ▼
┌─────────────────────────────────────────┐
│      Service Layer (Business Logic)     │
│  • customer_service.py                  │
│  • order_service.py                     │
│  • user_service.py                      │
│  ├─ Permission checks (require)         │
│  ├─ Validation (Pydantic)               │
│  └─ Audit logging                       │
└────────────────┬────────────────────────┘
                 │
                 ▼
┌─────────────────────────────────────────┐
│   Repository Layer (Data Access)        │
│  • customer_repository.py               │
│  • order_repository.py                  │
│  • user_repository.py                   │
│  └─ SQLAlchemy CRUD operations          │
└────────────────┬────────────────────────┘
                 │
                 ▼
┌─────────────────────────────────────────┐
│    SQLite Database (db.sqlite3)         │
│  • users (id, username, password, role) │
│  • customers (id, name, email, city)    │
│  • orders (id, customer_id, product...) │
└─────────────────────────────────────────┘

Database Schema

Users Table

CREATE TABLE users (
    id INTEGER PRIMARY KEY,
    username VARCHAR(50) UNIQUE,
    password VARCHAR(255),
    role VARCHAR(20),  -- admin, manager, viewer
    created_at DATETIME
)

Customers Table

CREATE TABLE customers (
    id INTEGER PRIMARY KEY,
    name VARCHAR(100),
    email VARCHAR(100) UNIQUE,
    city VARCHAR(100),
    created_at DATETIME,
    updated_at DATETIME
)

Orders Table

CREATE TABLE orders (
    id INTEGER PRIMARY KEY,
    customer_id INTEGER FOREIGN KEY,
    product VARCHAR(200),
    amount FLOAT,
    status VARCHAR(20),  -- Pending, Completed, Cancelled
    created_at DATETIME,
    updated_at DATETIME
)

Role-Based Permissions

Viewer Role

• Read-only access to all resources
• Permissions: customer.read, order.read, user.read
• Denied: All write and delete operations

Manager Role

• Create and update customers and orders
• Cannot delete resources
• Permissions: customer.read, customer.write, order.read, order.write, user.read
• Denied: Delete operations

Admin Role

• Full access to all resources
• Permissions: All operations

Installation

Prerequisites

• Python 3.12+
• pip or uv package manager

Setup

# Clone the repository
cd mcp-sqlite-rbac-demo

# Install dependencies
pip install -r requirements.txt
# or with uv:
uv pip install -r requirements.txt

# Seed the database with sample data
python seed.py

# Start the MCP server
python server.py

Sample Credentials

Admin:    username=admin,    password=admin123
Manager:  username=manager,  password=manager123
Viewer:   username=viewer,   password=viewer123

How MCP Works

Model Context Protocol (MCP) is a standardized interface for LLMs to interact with external tools and data. This server implements MCP by:

1. Tool Registration - Server exposes 20+ tools via MCP
2. JSON-RPC Communication - Tools are called via JSON-RPC protocol
3. Authentication - Tools enforce role-based access control
4. Structured Input/Output - All tools have schemas for type safety

MCP Tool Categories

Authentication Tools

• login(username, password) - Authenticate user
• logout() - End user session
• whoami() - Get current user info
• my_permissions() - Get user's permissions

Customer Tools

• list_customers(skip=0, limit=100) - List all customers
• get_customer(customer_id) - Get customer by ID
• search_customers(name, skip=0, limit=100) - Search by name
• create_customer(name, email, city) - Create new customer
• update_customer(customer_id, name?, email?, city?) - Update customer
• delete_customer(customer_id) - Delete customer

Order Tools

• list_orders(skip=0, limit=100) - List all orders
• get_order(order_id) - Get order by ID
• get_customer_orders(customer_id, skip=0, limit=100) - Get customer's orders
• list_orders_by_status(status, skip=0, limit=100) - Filter by status
• create_order(customer_id, product, amount) - Create new order
• update_order_status(order_id, status) - Update order status
• delete_order(order_id) - Delete order

User Tools

• list_users(skip=0, limit=100) - List all users
• get_user(user_id) - Get user by ID

Usage Examples

Example 1: Viewer Role (Read-Only)

# 1. Login as viewer
login(username="viewer", password="viewer123")
# Returns: {"username": "viewer", "role": "viewer", ...}

# 2. Check permissions
my_permissions()
# Returns: {"role": "viewer", "permissions": ["customer.read", "order.read", "user.read"]}

# 3. List customers (allowed)
list_customers()
# Returns: {"customers": [...], "count": 3}

# 4. Try to create customer (denied)
create_customer(name="Jane Doe", email="jane@test.com", city="LA")
# Returns: PermissionError: Permission 'customer.write' denied for role 'viewer'

Example 2: Manager Role (Create/Update)

# 1. Login as manager
login(username="manager", password="manager123")

# 2. Create new customer
create_customer(name="Jane Doe", email="jane@test.com", city="LA")
# Returns: {"id": 4, "name": "Jane Doe", ...}

# 3. Update customer
update_customer(customer_id=4, city="San Francisco")
# Returns: {"id": 4, "name": "Jane Doe", "city": "San Francisco", ...}

# 4. Try to delete customer (denied)
delete_customer(customer_id=4)
# Returns: PermissionError: Permission 'customer.delete' denied for role 'manager'

Example 3: Admin Role (Full Access)

# 1. Login as admin
login(username="admin", password="admin123")

# 2. List all users
list_users()
# Returns: {"users": [...], "count": 3}

# 3. Delete customer
delete_customer(customer_id=4)
# Returns: {"message": "Customer 4 deleted successfully"}

# 4. All operations allowed

Permission Model

The permission system uses a simple role-to-permission mapping:

PERMISSIONS = {
    "admin": {
        "user.read", "user.write",
        "customer.read", "customer.write", "customer.delete",
        "order.read", "order.write", "order.delete",
    },
    "manager": {
        "user.read",
        "customer.read", "customer.write",
        "order.read", "order.write",
    },
    "viewer": {
        "user.read",
        "customer.read",
        "order.read",
    },
}

Every write operation calls require(role, permission) which raises PermissionError if denied.

Audit Logging

Every operation is logged to audit.log with:

• Timestamp
• Username
• Action (e.g., CREATE_CUSTOMER, LOGIN)
• Resource (e.g., customer:123)
• Result (success or error)

{
  "timestamp": "2024-07-30T10:15:30.123456",
  "username": "admin",
  "action": "CREATE_CUSTOMER",
  "resource": "customer:4",
  "result": "success"
}

Integrating with AI Clients

Claude Desktop

Add to Claude Desktop config (~/.claude/claude_desktop_config.json):

{
  "mcpServers": {
    "sqlite-rbac": {
      "command": "python",
      "args": ["/path/to/mcp-sqlite-rbac-demo/server.py"]
    }
  }
}

Then start Claude Desktop and the server will be available.

OpenAI Agents SDK

import subprocess
from openai import OpenAI

# Start MCP server subprocess
server_process = subprocess.Popen([
    "python", "/path/to/server.py"
], stdout=subprocess.PIPE, stderr=subprocess.PIPE)

client = OpenAI(api_key="your-key")

# Use MCP server tools with OpenAI
response = client.beta.agents.create(
    name="Database Agent",
    tools=[
        # Tools from MCP server will be available
    ],
    model="gpt-4"
)

LangGraph

from langchain.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent
import subprocess
import json

# Start MCP server
server = subprocess.Popen(["python", "server.py"])

# Define tools that call MCP server
@tool
def list_customers():
    """List all customers"""
    # Call MCP server over stdin/stdout
    ...

# Create LangGraph agent
agent = create_react_agent(ChatOpenAI(model="gpt-4"), [list_customers])

Cursor AI Editor

Add MCP server to Cursor settings (.cursor/settings.json):

{
  "mcpServers": {
    "sqlite-rbac": {
      "command": "python",
      "args": ["server.py"]
    }
  }
}

Production Deployment

PostgreSQL Migration

To migrate from SQLite to PostgreSQL for production:

# 1. Update database URL
settings.database_url = "postgresql://user:password@localhost/rbac_db"

# 2. Change connection settings
engine = create_engine(
    settings.database_url,
    # Remove SQLite-specific options
    # No need for StaticPool with PostgreSQL
)

# 3. Install psycopg2
pip install psycopg2-binary

# 4. Run migrations
alembic upgrade head

# 5. Update connection pooling (optional)
from sqlalchemy.pool import QueuePool
engine = create_engine(
    settings.database_url,
    poolclass=QueuePool,
    pool_size=10,
    max_overflow=20,
)

Production Configuration

# config.py
class Settings(BaseSettings):
    database_url: str  # Read from environment
    debug: bool = False
    secret_key: str  # For session encryption
    log_level: str = "INFO"
    audit_log_retention_days: int = 90

Security Hardening

1. Password Hashing - Replace plaintext passwords with bcrypt:

from passlib.context import CryptContext
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
hashed = pwd_context.hash("password")

2. Session Encryption - Use secure tokens:

import secrets
session_token = secrets.token_urlsafe(32)

3. Rate Limiting - Limit login attempts:

from slowapi import Limiter
limiter = Limiter(key_func=get_remote_address)

4. HTTPS Only - Use TLS in production

5. SQL Injection Prevention - Already using SQLAlchemy ORM (safe)

Code Quality

• Type Hints - Full type annotations throughout
• Docstrings - Every function documented
• Error Handling - Friendly error messages
• Validation - Pydantic schemas for all inputs
• Clean Architecture - Separation of concerns
• DRY Principle - No code duplication

Testing

# Run example queries (manual testing)
python seed.py  # Populate database

# Test permission enforcement
python -c "
from auth import get_session
from services.customer_service import CustomerService

# Login as viewer
session = get_session()
session.login('viewer', 'viewer123')

# Try to create (should fail)
try:
    service = CustomerService()
    service.create_customer({'name': 'Test'}, session.get_role())
except PermissionError as e:
    print(f'✓ Permission denied as expected: {e}')
"

File Structure

mcp-sqlite-rbac-demo/
├── README.md                 # This file
├── requirements.txt          # Python dependencies
├── config.py                 # Configuration management
├── database.py               # SQLAlchemy setup
├── models.py                 # ORM models (User, Customer, Order)
├── schemas.py                # Pydantic request/response schemas
├── auth.py                   # Authentication and session management
├── permissions.py            # RBAC permission system
├── audit.py                  # Audit logging
├── seed.py                   # Database seeding script
├── server.py                 # Main MCP server
├── services/
│   ├── customer_service.py   # Customer business logic
│   ├── order_service.py      # Order business logic
│   └── user_service.py       # User business logic
├── repositories/
│   ├── customer_repository.py # Customer data access
│   ├── order_repository.py    # Order data access
│   └── user_repository.py     # User data access
├── tools/
│   ├── auth_tools.py         # MCP tools for auth
│   ├── customer_tools.py     # MCP tools for customers
│   ├── order_tools.py        # MCP tools for orders
│   └── user_tools.py         # MCP tools for users
├── db.sqlite3                # SQLite database (auto-created)
└── audit.log                 # Audit log entries

Key Concepts Demonstrated

1. Clean Architecture

• Tools Layer - MCP tool definitions with schemas
• Service Layer - Business logic and validation
• Repository Layer - Data access and ORM
• Database Layer - SQLAlchemy with relationships

2. Authentication & Authorization

• Session-based authentication (singleton pattern)
• Role-based access control with granular permissions
• Permission enforcement at service layer
• Audit logging for compliance

3. Type Safety

• Pydantic schemas for validation
• SQLAlchemy models for type-safe database access
• Full type hints throughout codebase

4. Error Handling

• Custom exceptions (PermissionError, ValueError)
• Friendly error messages
• Audit logging of failures
• Proper HTTP error codes

5. Enterprise Patterns

• Dependency injection (services accept db)
• Repository pattern for data access
• Service layer for business logic
• Pagination and filtering
• Relationship management

Common Patterns

Permission Check

from permissions import require

def create_customer(data, role):
    require(role, "customer.write")  # Raises PermissionError if denied
    # ... create customer

Service Method

def get_customer(self, customer_id, role):
    require(role, "customer.read")
    customer = self.repo.get_by_id(customer_id)
    if not customer:
        raise ValueError(f"Customer {customer_id} not found")
    return CustomerResponse.model_validate(customer)

MCP Tool

def list_customers(skip=0, limit=100):
    session = get_session()
    if not session.is_authenticated():
        raise RuntimeError("Not authenticated")
    
    service = CustomerService()
    customers = service.list_customers(session.get_role(), skip=skip, limit=limit)
    log_audit(session.get_username(), "LIST_CUSTOMERS", "customer", "success")
    return {"customers": [c.model_dump() for c in customers]}

Extending the Project

Add New Resource Type (e.g., Products)

1. Create Model (models.py):

class Product(Base):
    __tablename__ = "products"
    id = Column(Integer, primary_key=True)
    name = Column(String(100), nullable=False)
    price = Column(Float, nullable=False)

2. Create Repository (repositories/product_repository.py):

class ProductRepository:
    def __init__(self, db):
        self.db = db
    def get_all(self):
        return self.db.query(Product).all()

3. Create Service (services/product_service.py):

class ProductService:
    def __init__(self, db=None):
        self.repo = ProductRepository(db or SessionLocal())
    def list_products(self, role):
        require(role, "product.read")
        return self.repo.get_all()

4. Create Tools (tools/product_tools.py):

def list_products():
    session = get_session()
    if not session.is_authenticated():
        raise RuntimeError("Not authenticated")
    service = ProductService()
    products = service.list_products(session.get_role())
    return {"products": [p.model_dump() for p in products]}

5. Register in Server (server.py):

from tools.product_tools import get_product_tools
all_tools = [...] + get_product_tools()

Add New Role (e.g., Analyst)

1. Update Permissions (permissions.py):

PERMISSIONS = {
    # ...
    "analyst": {
        "customer.read",
        "order.read",
        "report.read",
    },
}

2. Seed Users (seed.py):

User(username="analyst", password="analyst123", role="analyst")

License

This project is provided as educational material for learning MCP server development.

Support

For questions or issues, refer to:

• MCP Documentation
• FastMCP SDK
• SQLAlchemy Docs
• Pydantic Docs

Troubleshooting

Database locked error

• SQLite can be slow with concurrent access
• For production, use PostgreSQL instead

Permission denied errors

• Verify you're logged in with whoami()
• Check your permissions with my_permissions()
• Use a higher-role account (manager, admin)

No tables in database

• Run python seed.py to initialize the database
• Check that db.sqlite3 was created

Tool not found

• Ensure all dependencies are installed: pip install -r requirements.txt
• Restart the server after modifying tools
• Check server logs for errors