symfonic.agent.fastapi.dependencies

dependencies

FastAPI dependencies for tenant scope extraction.

Provides get_tenant_scope as a FastAPI Depends function that extracts multi-tenant context from HTTP headers.

Security model

X-Tenant-ID on its own is untrusted input. The core library cannot assume a specific auth model (JWT, API key, mTLS, ...), so it exposes a pluggable async verifier that downstream apps register at startup to enforce "this caller is allowed to access this tenant".

Two forms exist, and an injected resolver wins over the global wherever both are present -- in this dependency and in the production gate, which have to agree about that or the gate validates something no request will use.

Preferred (what symfonic init generates, see templates/app/main.py.j2):

.. code-block:: python

from symfonic.agent.fastapi.dependencies import install_scope_resolver
from symfonic.platform import HeaderScopeResolver

async def verify_tenant_access(credentials, tenant_id):
    # decode JWT, look up user, confirm membership in tenant_id.
    # Raise AuthenticationError / AuthorizationError to deny.
    return {"principal_id": user.id, "is_admin": user.role == "admin"}

install_scope_resolver(app, HeaderScopeResolver(verifier=verify_tenant_access))
router = create_agent_router(agent, app=app)

Legacy, still supported (a process global holding a per-deployment fact):

.. code-block:: python

from symfonic.agent.fastapi.dependencies import set_tenant_auth_verifier

async def verify_tenant_access(request, tenant_id):
    ...  # raise HTTPException to deny

set_tenant_auth_verifier(verify_tenant_access)

When neither is wired the dependency logs a WARNING once and accepts X-Tenant-ID on trust -- suitable for tests / dev only, never production, which is what :func:_check_production_auth_gate refuses to let you ship.

TenantAuthVerifier module-attribute

TenantAuthVerifier = Callable[[Request, str], Awaitable[None]]

Signature: async def verifier(request, tenant_id) -> None.

Implementations should raise fastapi.HTTPException (401 / 403) to deny the request. Returning None is an explicit allow.

get_scope_resolver

get_scope_resolver(app: Any) -> Any | None

Return the resolver injected into app, if any.

Source code in symfonic/agent/fastapi/dependencies.py

def get_scope_resolver(app: Any) -> Any | None:
    """Return the resolver injected into ``app``, if any."""
    return getattr(app.state, SCOPE_RESOLVER_STATE_ATTR, None)

get_tenant_auth_verifier

get_tenant_auth_verifier() -> TenantAuthVerifier | None

Return the currently registered verifier, if any.

Exposed primarily for diagnostics / tests.

Source code in symfonic/agent/fastapi/dependencies.py

def get_tenant_auth_verifier() -> TenantAuthVerifier | None:
    """Return the currently registered verifier, if any.

    Exposed primarily for diagnostics / tests.
    """
    return _tenant_auth_verifier

get_tenant_scope async

get_tenant_scope(request: Request) -> FrameworkTenantScope

Extract a FrameworkTenantScope from request headers.

Required headers

X-Tenant-ID: Unique tenant identifier (required).

Optional headers

X-Sub-Tenant-ID: Sub-tenant identifier. X-Namespace: Memory namespace.

Security

If a tenant-auth verifier was registered via set_tenant_auth_verifier it is invoked with the request and the parsed tenant id. The verifier is expected to confirm the caller is authenticated AND authorised for the requested tenant, raising HTTPException(401/403) on denial.

When no verifier is registered a one-shot WARNING is logged and the tenant id is accepted on trust -- acceptable for unit tests and local dev, never for production.

Raises:

Type	Description
HTTPException 401	Missing X-Tenant-ID header, missing auth, or invalid credentials (propagated from the verifier).
HTTPException 403	Caller is authenticated but not a member of the requested tenant (propagated from the verifier).
HTTPException 400	Invalid scope (e.g. divergent sub_tenant_id/namespace).

Source code in symfonic/agent/fastapi/dependencies.py

async def get_tenant_scope(request: Request) -> FrameworkTenantScope:
    """Extract a FrameworkTenantScope from request headers.

    Required headers:
        X-Tenant-ID: Unique tenant identifier (required).

    Optional headers:
        X-Sub-Tenant-ID: Sub-tenant identifier.
        X-Namespace: Memory namespace.

    Security:
        If a tenant-auth verifier was registered via
        ``set_tenant_auth_verifier`` it is invoked with the request and
        the parsed tenant id.  The verifier is expected to confirm the
        caller is authenticated AND authorised for the requested tenant,
        raising ``HTTPException(401/403)`` on denial.

        When no verifier is registered a one-shot WARNING is logged and
        the tenant id is accepted on trust -- acceptable for unit tests
        and local dev, never for production.

    Raises:
        HTTPException 401: Missing X-Tenant-ID header, missing auth, or
            invalid credentials (propagated from the verifier).
        HTTPException 403: Caller is authenticated but not a member of
            the requested tenant (propagated from the verifier).
        HTTPException 400: Invalid scope (e.g. divergent
            sub_tenant_id/namespace).
    """
    # Before anything else: the app serving this request must satisfy the
    # production gate, whatever app was validated at construction time.
    _enforce_serving_app(request)

    resolver = get_scope_resolver(request.app) if hasattr(request, "app") else None
    if resolver is not None:
        return await _scope_via_resolver(request, resolver)

    tenant_id = request.headers.get("X-Tenant-ID")
    if not tenant_id:
        raise HTTPException(
            status_code=401,
            detail="Missing required X-Tenant-ID header",
        )

    # Enforce the pluggable auth hook *before* building the scope so
    # unauthorised callers never reach the memory orchestrator.
    global _missing_verifier_warning_emitted
    if _tenant_auth_verifier is not None:
        await _tenant_auth_verifier(request, tenant_id)
    elif not _missing_verifier_warning_emitted:
        logger.warning(
            "No tenant auth verifier registered — X-Tenant-ID is being "
            "trusted unconditionally. Call "
            "symfonic.agent.fastapi.dependencies.set_tenant_auth_verifier() "
            "at app startup to enforce authentication + tenant membership.",
        )
        _missing_verifier_warning_emitted = True

    sub_tenant_id = request.headers.get("X-Sub-Tenant-ID") or None
    namespace = request.headers.get("X-Namespace") or None

    try:
        return FrameworkTenantScope(
            tenant_id=tenant_id,
            sub_tenant_id=sub_tenant_id,
            namespace=namespace,
        )
    except (ScopeValidationError, ValueError) as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc

install_scope_resolver

install_scope_resolver(app: Any, resolver: Any) -> None

Inject the scope resolver this app derives principals with.

The preferred form. :func:set_tenant_auth_verifier registers a process global for a per-deployment fact, so two agents mounted in one process share one verifier and the second one to start silently redefines authorisation for the first. A resolver held on app.state is scoped to the app that owns it, which is the whole point of :class:~symfonic.platform.scope.HeaderScopeResolver.

resolver is anything satisfying :class:~symfonic.platform.ports.ScopeResolver -- async resolve( credentials) -> AuthenticatedPrincipal.

Note the two verifier protocols are not interchangeable. The legacy one takes (Request, tenant_id) and denies by raising HTTPException; the platform one takes (RequestCredentials, tenant_id) and returns principal facts. A legacy verifier therefore cannot be wrapped into a resolver without losing its access to the request, so this function does not try: an app either injects a resolver or keeps its verifier, and :func:get_tenant_scope honours whichever is present.

Parameters:

Name	Type	Description	Default
app	Any	The FastAPI application (anything with .state).	required
resolver	Any	The resolver, or None to clear it.	required

Source code in symfonic/agent/fastapi/dependencies.py

def install_scope_resolver(app: Any, resolver: Any) -> None:
    """Inject the scope resolver this app derives principals with.

    The preferred form. :func:`set_tenant_auth_verifier` registers a
    *process* global for a *per-deployment* fact, so two agents mounted in one
    process share one verifier and the second one to start silently redefines
    authorisation for the first. A resolver held on ``app.state`` is scoped to
    the app that owns it, which is the whole point of
    :class:`~symfonic.platform.scope.HeaderScopeResolver`.

    ``resolver`` is anything satisfying
    :class:`~symfonic.platform.ports.ScopeResolver` -- ``async resolve(
    credentials) -> AuthenticatedPrincipal``.

    Note the two verifier protocols are **not** interchangeable. The legacy one
    takes ``(Request, tenant_id)`` and denies by raising ``HTTPException``; the
    platform one takes ``(RequestCredentials, tenant_id)`` and *returns
    principal facts*. A legacy verifier therefore cannot be wrapped into a
    resolver without losing its access to the request, so this function does not
    try: an app either injects a resolver or keeps its verifier, and
    :func:`get_tenant_scope` honours whichever is present.

    Args:
        app: The ``FastAPI`` application (anything with ``.state``).
        resolver: The resolver, or ``None`` to clear it.
    """
    setattr(app.state, SCOPE_RESOLVER_STATE_ATTR, resolver)

set_tenant_auth_verifier

set_tenant_auth_verifier(verifier: TenantAuthVerifier | None) -> None

Register (or clear) the tenant-access verifier.

Call this once at app startup, before mounting the agent router. Passing None unregisters the current verifier (useful for tests).

The verifier receives the raw Request plus the parsed tenant_id from the X-Tenant-ID header and must raise an HTTPException to deny access. Any other exception is treated as a 500 by FastAPI.

Source code in symfonic/agent/fastapi/dependencies.py

def set_tenant_auth_verifier(verifier: TenantAuthVerifier | None) -> None:
    """Register (or clear) the tenant-access verifier.

    Call this once at app startup, *before* mounting the agent router.
    Passing ``None`` unregisters the current verifier (useful for tests).

    The verifier receives the raw ``Request`` plus the parsed ``tenant_id``
    from the ``X-Tenant-ID`` header and must raise an ``HTTPException`` to
    deny access.  Any other exception is treated as a 500 by FastAPI.
    """
    global _tenant_auth_verifier, _missing_verifier_warning_emitted
    _tenant_auth_verifier = verifier
    # Reset the one-shot warning flag so re-registering after a clear
    # doesn't silently swallow the dev-mode warning again.
    _missing_verifier_warning_emitted = False