app.dependencies

FastAPI dependencies for Kumiho integration.

Functions

ensure_tenant_context()	Return tenant context from environment service token.
get_kumiho_client(request[, x_kumiho_token, ...])	Creates a Kumiho client instance for the current request using the provided token.
get_user_token([authorization])	Extract Firebase user token from Authorization header (optional).
require_user_token([token])	Require a Firebase user token in the Authorization header.

Classes

TenantContext(auth_token, expires_at, ...)

Cached tenant/control-plane metadata derived from the CP JWT.

class app.dependencies.TenantContext(auth_token, expires_at, tenant_id, tenant_slug, tenant_tier, region_code, neo4j_db_name, guardrails, roles, server_url, grpc_authority)

Bases: object

Cached tenant/control-plane metadata derived from the CP JWT.

Parameters:

• auth_token (str)

• expires_at (float)

• tenant_id (str)

• tenant_slug (str | None)

• tenant_tier (str | None)

• region_code (str | None)

• neo4j_db_name (str | None)

• guardrails (Dict[str, Any])

• roles (Tuple[str, ...])

• server_url (str | None)

• grpc_authority (str | None)

auth_token: str

expires_at: float

tenant_id: str

tenant_slug: str | None

tenant_tier: str | None

region_code: str | None

neo4j_db_name: str | None

guardrails: Dict[str, Any]

roles: Tuple[str, ...]

server_url: str | None

grpc_authority: str | None

preferred_target()

Return the best known endpoint for the tenant.

Return type:

Optional[str]

app.dependencies.ensure_tenant_context()

Return tenant context from environment service token.

Return type:

TenantContext

app.dependencies.get_user_token(authorization=Header(None))

Extract Firebase user token from Authorization header (optional).

This is for end-user authentication that gets passed through to kumiho-server. The service token (X-Kumiho-Token) is still required for API access.

Parameters:

authorization (Optional[str]) – Optional Authorization header with Bearer token

Return type:

Optional[str]

Returns:

The extracted token or None

async app.dependencies.get_kumiho_client(request, x_kumiho_token=Header(None), x_tenant_id=Header(None), x_correlation_id=Header(None), x_idempotency_key=Header(None), user_token=Depends(dependency=<function get_user_token>, use_cache=True, scope=None))

Creates a Kumiho client instance for the current request using the provided token.

This is a per-request client - each API call gets its own client instance configured with the client’s service token. This enables multi-tenant operation where different clients can use the same FastAPI deployment with their own tokens.

Parameters:

• x_kumiho_token (Optional[str]) – The client’s Kumiho service token from X-Kumiho-Token header

• x_tenant_id (Optional[str]) – Optional tenant ID hint for anonymous users

• user_token (Optional[str]) – Optional user ID token from Authorization: Bearer header

• request (Request)

• x_correlation_id (str | None)

• x_idempotency_key (str | None)

Returns:

A configured kumiho.Client instance

Raises:

• HTTPException – 401 if token is missing or invalid

• HTTPException – 500 if client creation fails

app.dependencies.require_user_token(token=Header(None))

Require a Firebase user token in the Authorization header.

Parameters:

token (Optional[str]) – Authorization header value

Return type:

str

Returns:

The extracted Firebase token

Raises:

HTTPException – 401 if token is missing or invalid format