Protect a FastMCP App

FastMCP accepts an auth callable in its Server constructor. CaracalAuth from caracalai_mcp_fastmcp is a callable that takes a bearer token string and returns the validated Claims on success or raises CaracalAuthError on failure. FastMCP calls it with the token extracted from the Authorization header before invoking any tool handler.

Prerequisites

A FastMCP server.
The STS JWKS endpoint reachable from the Python process.
pip install fastmcp (FastMCP is not bundled with the Caracal connector).

Install

pip install caracalai-mcp-fastmcp caracalai-revocation

Attach auth to a FastMCP server

from fastmcp.server import Server
from caracalai_mcp_fastmcp import CaracalAuth
from caracalai_revocation import InMemoryRevocationStore

revocations = InMemoryRevocationStore()

auth = CaracalAuth(
    issuer='http://sts:8080',
    audience='resource://my-mcp-server',
    revocations=revocations,
    required_scopes=['tool:call'],
    expected_zone_id='my-zone',
    require_agent=True,
)

server = Server('my-server', auth=auth)

Every tool registered on this server is now protected. Calls without a valid mandate are rejected before any tool code runs.

Register tools

Tool handlers receive the FastMCP context. The authenticated claims are not automatically injected into the handler argument list; access them via the context if FastMCP exposes it, or read them from the lifespan state you set up:

@server.tool()
async def search_documents(query: str, limit: int = 10) -> list[dict]:
    # Mandate is already verified at this point
    results = await document_store.search(query, limit=limit)
    return results

@server.tool()
async def submit_order(order_id: str, amount: float) -> dict:
    # Only callers with 'tool:call' scope reach here
    result = await orders.submit(order_id, amount)
    return {'submitted': True, 'result': result}

Auth options

Parameter	Type	Required	Description
`issuer`	`str`	Yes	STS base URL; JWKS at `{issuer}/.well-known/jwks.json`
`audience`	`str`	Yes	Must match the `aud` claim in the mandate
`revocations`	`RevocationStore`	Yes	Checked against the `sid` claim
`required_scopes`	`list[str]`	No	All listed scopes must be in the mandate
`expected_zone_id`	`str`	No	Rejects mandates from a different zone
`require_agent`	`bool`	No	Rejects mandates without `agent_session_id`
`require_delegation`	`bool`	No	Rejects mandates without `delegation_edge_id`
`require_chain_contains`	`list[str]`	No	All listed application IDs must be in `delegation_chain`
`max_hop_count`	`int`	No	Rejects mandates where `hop_count` exceeds this value

Error handling

CaracalAuth.__call__ raises CaracalAuthError on any verification failure. FastMCP maps this to an appropriate error response. The error has two attributes:

class CaracalAuthError(Exception):
    code: str        # e.g. 'insufficient_scope', 'session_revoked'
    description: str # Human-readable explanation

Possible code values: missing_token, invalid_token, invalid_zone, insufficient_scope, session_revoked, agent_required, delegation_required, chain_mismatch, hop_count_exceeded.

Production revocation

Replace InMemoryRevocationStore with the Redis-backed store and subscribe to the revocation stream:

import redis
from caracalai_revocation_redis import RedisRevocationStore, RedisRevocationConsumer
import threading

r = redis.Redis.from_url(os.environ['REDIS_URL'])
revocations = RedisRevocationStore(redis=r, fail_closed=True)

consumer = RedisRevocationConsumer(
    redis=r,
    store=revocations,
    consumer='fastmcp-server-1',
)
consumer.ensure_group()

def poll_revocations():
    while True:
        consumer.poll_once()

threading.Thread(target=poll_revocations, daemon=True).start()

auth = CaracalAuth(
    issuer=os.environ['CARACAL_STS_URL'],
    audience='resource://my-mcp-server',
    revocations=revocations,
    required_scopes=['tool:call'],
)

server = Server('my-server', auth=auth)

Complete example

import os
import asyncio
from fastmcp.server import Server
from caracalai_mcp_fastmcp import CaracalAuth
from caracalai_revocation import InMemoryRevocationStore

revocations = InMemoryRevocationStore()

auth = CaracalAuth(
    issuer=os.environ['CARACAL_STS_URL'],
    audience='resource://documents',
    revocations=revocations,
    required_scopes=['tool:call'],
    expected_zone_id=os.environ['CARACAL_ZONE_ID'],
    require_agent=True,
)

server = Server('documents-mcp', auth=auth)

@server.tool()
async def search(query: str) -> list[dict]:
    return [{'id': '1', 'title': f'Result for: {query}'}]

@server.tool()
async def get_document(doc_id: str) -> dict:
    return {'id': doc_id, 'content': 'Document content here'}

if __name__ == '__main__':
    asyncio.run(server.run())