Storage backends

Three pluggable backends share the same abstract interface.

Key management

file and redis backends encrypt all stored data with a Fernet symmetric key. The key is resolved in this order:

1. STORAGE_ENCRYPTION_KEY env var — a base64-encoded Fernet key (44-char URL-safe string)
2. STORAGE_ENCRYPTION_KEY_PATH env var — path to a file containing the key
3. Auto-generated ephemeral key — tokens are lost on restart; a warning is logged

Generating a stable key

python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())"

Store the output and export it before starting the server:

export STORAGE_ENCRYPTION_KEY=<paste key here>

Key rotation / wrong key

If the server starts with a different key than the one used to encrypt existing entries, decryption will fail silently: the stale entry is deleted and the user is re-prompted for credentials once. No data is corrupted and no error is surfaced to the user — they just re-authenticate.

This means key rotation requires no migration: restart with the new key and users re-auth on their next tool call.

Interfaces

base

Abstract base classes for the two store types used by auth providers.

TokenStore

Bases: ABC

Persistent, encrypted store for user tokens and credentials.

Keyed by the primary OIDC sub claim. Values are arbitrary JSON-serialisable dicts (TokenData for OAuth, or {field: value} for credentials providers).

get abstractmethod async

get(sub: str) -> dict | None

Return the stored dict for sub, or None if absent.

Source code in mcpauthkit/store/base.py

@abstractmethod
async def get(self, sub: str) -> dict | None:
    """Return the stored dict for *sub*, or ``None`` if absent."""

set abstractmethod async

set(sub: str, value: dict) -> None

Persist value for sub, overwriting any existing entry.

Source code in mcpauthkit/store/base.py

@abstractmethod
async def set(self, sub: str, value: dict) -> None:
    """Persist *value* for *sub*, overwriting any existing entry."""

delete abstractmethod async

delete(sub: str) -> None

Remove the entry for sub (no-op if absent).

Source code in mcpauthkit/store/base.py

@abstractmethod
async def delete(self, sub: str) -> None:
    """Remove the entry for *sub* (no-op if absent)."""

PendingStore

Bases: ABC

Ephemeral, encrypted store for in-flight elicitation state.

Each entry has a short-lived opaque key (the OAuth state parameter or the credential-form entry token) and carries only JSON-serialisable metadata (sub, expires_at, …). Non-serialisable objects (asyncio.Event, MCP sessions) are kept in the provider's local _sessions dict.

Signal / wait semantics

The callback handler calls set_result once the token or credentials are ready. The decorator that is waiting for that outcome calls wait_for_result to block until the result arrives or the timeout expires.

• memory — backed by asyncio.Event; zero-overhead wait.
• file / redis — backed by polling at 0.5 s intervals; fully adequate for human-interactive flows.

create abstractmethod async

create(key: str, metadata: dict, ttl: int) -> None

Create a new pending entry that expires in ttl seconds.

Source code in mcpauthkit/store/base.py

@abstractmethod
async def create(self, key: str, metadata: dict, ttl: int) -> None:
    """Create a new pending entry that expires in *ttl* seconds."""

get abstractmethod async

get(key: str) -> dict | None

Return the metadata for key, or None if absent / expired.

Source code in mcpauthkit/store/base.py

@abstractmethod
async def get(self, key: str) -> dict | None:
    """Return the metadata for *key*, or ``None`` if absent / expired."""

pop abstractmethod async

pop(key: str) -> dict | None

Return and atomically delete the metadata for key.

Source code in mcpauthkit/store/base.py

@abstractmethod
async def pop(self, key: str) -> dict | None:
    """Return and atomically delete the metadata for *key*."""

set_result abstractmethod async

set_result(key: str, result: dict, ttl: int = 120) -> None

Store the completion result for key and wake any waiter.

Called by the instance that received the OAuth callback or the credential form submission. ttl controls how long the result is kept (important for file / redis; in-memory it is discarded after the waiter consumes it).

Source code in mcpauthkit/store/base.py

@abstractmethod
async def set_result(self, key: str, result: dict, ttl: int = 120) -> None:
    """
    Store the completion result for *key* and wake any waiter.

    Called by the instance that received the OAuth callback or the
    credential form submission.  ``ttl`` controls how long the result
    is kept (important for file / redis; in-memory it is discarded
    after the waiter consumes it).
    """

wait_for_result abstractmethod async

wait_for_result(key: str, timeout: float) -> dict | None

Block until a result is available for key, then return it.

Returns None on timeout. Implementations MUST consume (delete) the result entry so it is not returned twice.

Source code in mcpauthkit/store/base.py

@abstractmethod
async def wait_for_result(self, key: str, timeout: float) -> dict | None:
    """
    Block until a result is available for *key*, then return it.

    Returns ``None`` on timeout.  Implementations MUST consume (delete)
    the result entry so it is not returned twice.
    """

In-process (memory)

memory

In-memory store implementations (single-process, no persistence).

MemoryTokenStore — plain dict, keyed by OIDC sub. MemoryPendingStore — dict + asyncio.Event for zero-overhead signalling.

State is lost on server restart. No encryption — data lives only in process memory. Suitable for single-process deployments.

MemoryTokenStore

MemoryTokenStore()

Bases: TokenStore

Plain in-memory token / credential store.

Source code in mcpauthkit/store/memory.py

def __init__(self) -> None:
    self._data: dict[str, dict] = {}

MemoryPendingStore

MemoryPendingStore()

Bases: PendingStore

In-memory pending store.

Two internal dicts keep metadata and completion results separate so that pop (which removes the pending entry) and set_result (which signals the waiter) are fully decoupled:

_pending key → {metadata, _expires} _done key → {asyncio.Event, result}

Source code in mcpauthkit/store/memory.py

def __init__(self) -> None:
    self._pending: dict[str, dict] = {}
    self._done: dict[str, dict] = {}

File (Fernet-encrypted)

file_store

Encrypted file-based store implementations.

Suitable for multiple processes on the same host that share a filesystem (e.g. several uvicorn workers behind a reverse proxy, or replicas using a shared NFS / EFS volume).

Each entry is stored as a Fernet-encrypted JSON blob in its own file. Atomic writes use the tmp-then-rename pattern (POSIX-safe).

Directory layout::

{FILE_STORAGE_PATH}/
    tokens/
        {sha256(sub)[:16]}.enc          ← token / credential data
    pending/
        {sha256(key)[:16]}.enc          ← pending-flow metadata
        {sha256(key)[:16]}.done.enc     ← completion result (ephemeral)

Poll interval for wait_for_result: 0.5 s — more than fast enough for human-interactive OAuth / credential flows.

FileTokenStore

FileTokenStore(
    storage_path: str, namespace: str | None = None
)

Bases: TokenStore

Encrypted file-based token / credential store.

Each user's data is a separate .enc file named by the SHA-256 of their OIDC sub, stored under {storage_path}/tokens/{namespace}/. When namespace is omitted the subdirectory is simply tokens/.

Source code in mcpauthkit/store/file_store.py

def __init__(self, storage_path: str, namespace: str | None = None) -> None:
    base = Path(storage_path) / "tokens"
    self._dir = base / namespace if namespace else base
    self._dir.mkdir(parents=True, exist_ok=True)

FilePendingStore

FilePendingStore(storage_path: str)

Bases: PendingStore

Encrypted file-based pending store with polling-based wait_for_result.

Completion results are written to a separate .done.enc sidecar file so that pop (removes the pending entry) and set_result (writes the completion file) are fully independent operations.

Source code in mcpauthkit/store/file_store.py

def __init__(self, storage_path: str) -> None:
    self._dir = Path(storage_path) / "pending"
    self._dir.mkdir(parents=True, exist_ok=True)

Redis (async)

redis_store

Encrypted Redis-based store implementations.

Suitable for distributed / cloud deployments (multiple hosts or containers). All values are Fernet-encrypted before being written to Redis, so the Redis server never sees plaintext tokens or credentials.

Requires redis[asyncio] >= 5::

pip install "redis[asyncio]>=5"

Key layout ({prefix} defaults to mcp:auth:)::

{prefix}token:{sha256(sub)}    ← encrypted token / credential data
{prefix}pending:{sha256(key)}  ← encrypted pending-flow metadata (TTL set)
{prefix}done:{sha256(key)}     ← encrypted completion result (short TTL)

Poll interval for wait_for_result: 0.5 s.

RedisTokenStore

RedisTokenStore(redis_client, prefix: str = 'mcp:auth:')

Bases: TokenStore

Encrypted Redis token / credential store.

Token expiry is managed lazily by the provider (same behaviour as MemoryTokenStore) — no Redis TTL is set on token keys.

Source code in mcpauthkit/store/redis_store.py

def __init__(self, redis_client, prefix: str = "mcp:auth:") -> None:
    self._r = redis_client
    self._prefix = prefix

RedisPendingStore

RedisPendingStore(redis_client, prefix: str = 'mcp:auth:')

Bases: PendingStore

Encrypted Redis pending store with polling-based wait_for_result.

Completion results live under a separate done: key with a short TTL so that pop and set_result are fully decoupled operations, matching the semantics of the file and memory backends.

Source code in mcpauthkit/store/redis_store.py

def __init__(self, redis_client, prefix: str = "mcp:auth:") -> None:
    self._r = redis_client
    self._prefix = prefix

Factory

factory

Store factory — creates a (TokenStore, PendingStore) pair from configuration.

Reads TOKEN_STORAGE_MODE (and related env vars) unless overrides are passed explicitly. Call once at startup and inject the pair into every provider that needs storage.

Example::

from mcpauthkit.store import create_stores

token_store, pending_store = create_stores()

github_oauth = OAuthProvider.from_standard_oauth2(
    ...,
    token_store=token_store,
    pending_store=pending_store,
)

create_stores

create_stores(
    *,
    mode: str | None = None,
    file_path: str | None = None,
    redis_url: str | None = None,
    redis_prefix: str | None = None,
    namespace: str | None = None,
) -> tuple[TokenStore, PendingStore]

Return a (TokenStore, PendingStore) pair for the requested mode.

Parameters (all optional — fall back to environment variables)

mode "memory" | "file" | "redis". Overrides TOKEN_STORAGE_MODE env var (default: "memory"). file_path Root directory for file storage. Overrides FILE_STORAGE_PATH env var (default: /tmp/mcp-auth-store). redis_url Redis connection URL. Overrides REDIS_URL env var (default: redis://localhost:6379/0). redis_prefix Redis key prefix. Overrides REDIS_KEY_PREFIX env var (default: mcp:auth:). namespace Optional provider namespace. For file mode, creates a subdirectory inside tokens/ so different providers don't share filenames. For Redis mode, appended to the key prefix (e.g. mcp:auth:github:).

Source code in mcpauthkit/store/factory.py

def create_stores(
    *,
    mode: str | None = None,
    file_path: str | None = None,
    redis_url: str | None = None,
    redis_prefix: str | None = None,
    namespace: str | None = None,
) -> tuple[TokenStore, PendingStore]:
    """
    Return a ``(TokenStore, PendingStore)`` pair for the requested mode.

    Parameters (all optional — fall back to environment variables)
    -------------------------------------------------------------
    mode
        ``"memory"`` | ``"file"`` | ``"redis"``.
        Overrides ``TOKEN_STORAGE_MODE`` env var (default: ``"memory"``).
    file_path
        Root directory for file storage.
        Overrides ``FILE_STORAGE_PATH`` env var (default: ``/tmp/mcp-auth-store``).
    redis_url
        Redis connection URL.
        Overrides ``REDIS_URL`` env var (default: ``redis://localhost:6379/0``).
    redis_prefix
        Redis key prefix.
        Overrides ``REDIS_KEY_PREFIX`` env var (default: ``mcp:auth:``).
    namespace
        Optional provider namespace. For file mode, creates a subdirectory
        inside ``tokens/`` so different providers don't share filenames.
        For Redis mode, appended to the key prefix (e.g. ``mcp:auth:github:``).
    """
    resolved_mode = (mode or os.environ.get("TOKEN_STORAGE_MODE", "memory")).lower().strip()
    logger.info("Token storage mode: %s", resolved_mode)

    if resolved_mode == "memory":
        return MemoryTokenStore(), MemoryPendingStore()

    if resolved_mode == "file":
        _require_encryption_key(resolved_mode)
        from .file_store import FilePendingStore, FileTokenStore

        path = file_path or os.environ.get("FILE_STORAGE_PATH", "/tmp/mcp-auth-store")
        logger.info("File storage path: %s  namespace: %s", path, namespace or "(none)")
        return FileTokenStore(path, namespace=namespace), FilePendingStore(path)

    if resolved_mode == "redis":
        _require_encryption_key(resolved_mode)
        try:
            import redis.asyncio as aioredis
        except ImportError as exc:
            raise ImportError(
                "TOKEN_STORAGE_MODE=redis requires 'redis[asyncio]>=5'. "
                "Install it with:  pip install 'redis[asyncio]>=5'"
            ) from exc

        from .redis_store import RedisPendingStore, RedisTokenStore

        url = redis_url or os.environ.get("REDIS_URL", "redis://localhost:6379/0")
        prefix = redis_prefix or os.environ.get("REDIS_KEY_PREFIX", "mcp:auth:")
        if namespace:
            prefix = f"{prefix.rstrip(':')}:{namespace}:"
        logger.info("Redis URL: %s  key prefix: %s", url, prefix)
        client = aioredis.from_url(url, decode_responses=False)
        return RedisTokenStore(client, prefix), RedisPendingStore(client, prefix)

    raise ValueError(
        f"Unknown TOKEN_STORAGE_MODE {resolved_mode!r}. Valid values: memory, file, redis"
    )