# 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
```sql
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
```sql
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
```sql
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

```bash
# 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)

```bash
# 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)

```bash
# 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)

```bash
# 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:

```python
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)

```json
{
  "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`):

```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

```python
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

```python
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`):

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

## Production Deployment

### PostgreSQL Migration

To migrate from SQLite to PostgreSQL for production:

```python
# 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

```python
# 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:
```python
from passlib.context import CryptContext
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
hashed = pwd_context.hash("password")
```

2. **Session Encryption** - Use secure tokens:
```python
import secrets
session_token = secrets.token_urlsafe(32)
```

3. **Rate Limiting** - Limit login attempts:
```python
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

```bash
# 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
```python
from permissions import require

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

### Service Method
```python
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
```python
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`):
```python
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`):
```python
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`):
```python
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`):
```python
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`):
```python
from tools.product_tools import get_product_tools
all_tools = [...] + get_product_tools()
```

### Add New Role (e.g., Analyst)

1. **Update Permissions** (`permissions.py`):
```python
PERMISSIONS = {
    # ...
    "analyst": {
        "customer.read",
        "order.read",
        "report.read",
    },
}
```

2. **Seed Users** (`seed.py`):
```python
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](https://modelcontextprotocol.io)
- [FastMCP SDK](https://github.com/jlowin/FastMCP)
- [SQLAlchemy Docs](https://docs.sqlalchemy.org)
- [Pydantic Docs](https://docs.pydantic.dev)

## 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