graph TD
    A[🤖 AI Agent<br/>Claude / Any LLM] -->|MCP Protocol| B

    subgraph MCP Gateway — Governed Tool Access
        B[Request Received] --> C{Layer 1<br/>Permission Check}
        C -->|Missing permissions| D[❌ Denied<br/>Audit logged]
        C -->|Permitted| E{Layer 2<br/>Blast-Radius Guard}
        E -->|HIGH risk, no confirmation| F[❌ Blocked<br/>Audit logged]
        E -->|Confirmed or LOW/MED| G{Layer 3<br/>Input Validation}
        G -->|Path traversal / SQL injection| H[❌ Blocked<br/>Audit logged]
        G -->|Clean inputs| I[✅ Tool Handler Executes]
    end

    I --> J[(Tool Registry<br/>name · permissions · risk_level)]
    I --> K[📋 Audit Trail<br/>every call · permitted or denied]

    subgraph Tools
        I --> L[📊 Read Metrics]
        I --> M[🔍 Query Database]
        I --> N[🚀 Trigger Rollback<br/>HIGH RISK — requires confirmed=True]
    end

Agent (Claude / any LLM)
        │
        ▼ MCP Protocol
┌─────────────────────────────┐
│        MCP Server           │
│  ┌──────────────────────┐   │
│  │    Guardrail Layer   │   │  ← permission check → blast-radius guard → arg validation
│  └──────────┬───────────┘   │
│             │               │
│  ┌──────────▼───────────┐   │
│  │    Tool Registry     │   │  ← name, description, required_permissions, risk_level
│  └──────────┬───────────┘   │
│             │               │
│  ┌──────────▼───────────┐   │
│  │    Tool Handlers     │   │  ← plain Python functions, no security logic here
│  └──────────────────────┘   │
│             │               │
│  ┌──────────▼───────────┐   │
│  │     Audit Trail      │   │  ← every invocation logged, permitted or denied
│  └──────────────────────┘   │
└─────────────────────────────┘

registry.register(ToolDefinition(
    name="trigger_rollback",
    description="Initiate a deployment rollback.",
    handler=trigger_rollback,
    required_permissions={"deployments:write", "deployments:rollback"},
    risk_level=RiskLevel.HIGH,
    requires_confirmation=True,  # blast-radius guard
))

# Agent tries to trigger rollback but lacks deployments:write
guardrails.invoke(
    tool_name="trigger_rollback",
    arguments={"deployment_id": "d-123", "reason": "high error rate"},
    caller_id="monitoring-agent",
    caller_permissions={"deployments:read"},  # missing write permission
)
# → PermissionDeniedError: Caller 'monitoring-agent' lacks permissions
#   {'deployments:write', 'deployments:rollback'} for tool 'trigger_rollback'

# Without confirmation — blocked
guardrails.invoke("trigger_rollback", {...}, confirmed=False)
# → GuardrailViolationError: HIGH risk tool requires confirmed=True

# With confirmation — permitted
guardrails.invoke("trigger_rollback", {...}, confirmed=True)

# Path traversal — blocked automatically
guardrails.invoke("read_file", {"path": "../../etc/passwd"}, ...)
# → GuardrailViolationError: Path traversal detected

# Destructive SQL — blocked automatically
guardrails.invoke("query", {"query": "DROP TABLE users"}, ...)
# → GuardrailViolationError: Destructive SQL pattern detected

# After some invocations
events = audit.get_events()
print(events[0].to_json())
# {
#   "tool_name": "read_deployment_status",
#   "caller_id": "oncall-agent-v1",
#   "arguments": {"deployment_id": "d-abc"},
#   "result": "{'status': 'healthy', ...}",
#   "permitted": true,
#   "timestamp": "2026-08-26T14:30:00+00:00",
#   "duration_ms": 12.4
# }

print(f"Denied requests: {audit.denied_count()}")

production-mcp-server/
├── src/
│   ├── server.py          # MCP server entry point — tool registration + FastMCP wiring
│   ├── registry.py        # Tool registry — metadata, permissions, risk classification
│   ├── guardrails.py      # Guardrail layer — 3-layer enforcement on every invocation
│   ├── audit.py           # Structured audit trail — append-only event log
│   └── tools/
│       └── example_tools.py  # Example handlers — swap with your real data sources
├── tests/
│   ├── test_guardrails.py    # Permission enforcement, blast-radius, input validation
│   └── test_registry.py      # Tool registration and lookup
├── examples/
│   └── basic_usage.py        # Standalone usage without the MCP server
└── pyproject.toml

pip install -e ".[dev]"

python -m src.server

pytest tests/ -v

def read_config(config_key: str) -> str:
    return os.environ.get(config_key, "not_found")

registry.register(ToolDefinition(
    name="read_config",
    description="Read a configuration value by key.",
    handler=read_config,
    required_permissions={"config:read"},
    risk_level=RiskLevel.LOW,
))

@mcp.tool()
def config(config_key: str) -> str:
    return guardrails.invoke("read_config", {"config_key": config_key}, ...)

# Example: derive permissions from an OAuth token in the MCP session context
def get_caller_context(session) -> tuple[str, set[str]]:
    token = session.headers.get("Authorization")
    claims = verify_jwt(token)
    return claims["sub"], set(claims["permissions"])