Provides secure role-based access to a SQLite database, enabling AI agents to perform CRUD operations on customers and orders with authentication, permissions, and audit logging.
Click on "Install Server".
Wait a few minutes for the server to deploy. Once ready, it will show a "Started" state.
In the chat, type @ followed by the MCP server name and your instructions, e.g., "@MCP SQLite RBAC Demo show me all pending orders"
That's it! The server will respond to your query, and you can continue using it as needed.
Here is a step-by-step guide with screenshots.
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).
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
Related MCP server: sqlite-mcp
┌─────────────────────────────────────────┐
│ 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...) │
└─────────────────────────────────────────┘CREATE TABLE users (
id INTEGER PRIMARY KEY,
username VARCHAR(50) UNIQUE,
password VARCHAR(255),
role VARCHAR(20), -- admin, manager, viewer
created_at DATETIME
)CREATE TABLE customers (
id INTEGER PRIMARY KEY,
name VARCHAR(100),
email VARCHAR(100) UNIQUE,
city VARCHAR(100),
created_at DATETIME,
updated_at DATETIME
)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
)Read-only access to all resources
Permissions: customer.read, order.read, user.read
Denied: All write and delete operations
Create and update customers and orders
Cannot delete resources
Permissions: customer.read, customer.write, order.read, order.write, user.read
Denied: Delete operations
Full access to all resources
Permissions: All operations
Python 3.12+
pip or uv package manager
# 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.pyAdmin: username=admin, password=admin123
Manager: username=manager, password=manager123
Viewer: username=viewer, password=viewer123Model Context Protocol (MCP) is a standardized interface for LLMs to interact with external tools and data. This server implements MCP by:
Tool Registration - Server exposes 20+ tools via MCP
JSON-RPC Communication - Tools are called via JSON-RPC protocol
Authentication - Tools enforce role-based access control
Structured Input/Output - All tools have schemas for type safety
login(username, password) - Authenticate user
logout() - End user session
whoami() - Get current user info
my_permissions() - Get user's permissions
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
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
list_users(skip=0, limit=100) - List all users
get_user(user_id) - Get user by ID
# 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'# 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'# 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 allowedThe 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.
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"
}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.
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"
)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])Add MCP server to Cursor settings (.cursor/settings.json):
{
"mcpServers": {
"sqlite-rbac": {
"command": "python",
"args": ["server.py"]
}
}
}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,
)# 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 = 90Password Hashing - Replace plaintext passwords with bcrypt:
from passlib.context import CryptContext
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
hashed = pwd_context.hash("password")Session Encryption - Use secure tokens:
import secrets
session_token = secrets.token_urlsafe(32)Rate Limiting - Limit login attempts:
from slowapi import Limiter
limiter = Limiter(key_func=get_remote_address)HTTPS Only - Use TLS in production
SQL Injection Prevention - Already using SQLAlchemy ORM (safe)
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
# 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}')
"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 entriesTools Layer - MCP tool definitions with schemas
Service Layer - Business logic and validation
Repository Layer - Data access and ORM
Database Layer - SQLAlchemy with relationships
Session-based authentication (singleton pattern)
Role-based access control with granular permissions
Permission enforcement at service layer
Audit logging for compliance
Pydantic schemas for validation
SQLAlchemy models for type-safe database access
Full type hints throughout codebase
Custom exceptions (PermissionError, ValueError)
Friendly error messages
Audit logging of failures
Proper HTTP error codes
Dependency injection (services accept db)
Repository pattern for data access
Service layer for business logic
Pagination and filtering
Relationship management
from permissions import require
def create_customer(data, role):
require(role, "customer.write") # Raises PermissionError if denied
# ... create customerdef 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)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]}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)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()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()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]}Register in Server (server.py):
from tools.product_tools import get_product_tools
all_tools = [...] + get_product_tools()Update Permissions (permissions.py):
PERMISSIONS = {
# ...
"analyst": {
"customer.read",
"order.read",
"report.read",
},
}Seed Users (seed.py):
User(username="analyst", password="analyst123", role="analyst")This project is provided as educational material for learning MCP server development.
For questions or issues, refer to:
SQLite can be slow with concurrent access
For production, use PostgreSQL instead
Verify you're logged in with whoami()
Check your permissions with my_permissions()
Use a higher-role account (manager, admin)
Run python seed.py to initialize the database
Check that db.sqlite3 was created
Ensure all dependencies are installed: pip install -r requirements.txt
Restart the server after modifying tools
Check server logs for errors
This server cannot be installed
Unclaimed servers have limited discoverability.
If you are the server author, to access and configure the admin panel.
MCP server for Argo RPG Platform — connects AI assistants to campaign data via OAuth2
MCP server for AI agents to plan, verify, and deploy Cloudflare-native apps.
GibsonAI MCP server: manage your databases with natural language
We provide all the information about MCP servers via our MCP API.
curl -X GET 'https://glama.ai/api/mcp/v1/servers/phpdev-expert/mcp-server'
If you have feedback or need assistance with the MCP directory API, please join our Discord server