`aaa_server`¶

Package aaa-server; see the server overview and guide. Library docs: AAA home.

ServerConfig `dataclass` ¶

ServerConfig(
    engine: Engine,
    authenticator: Authenticator,
    algorithms: Mapping[str, Algorithm] = dict(),
    sessions: Mapping[str, SessionManager] = dict(),
    budget: Budget = (
        lambda: Budget(
            max_steps=50,
            max_tokens=200000,
            deadline_s=300.0,
        )
    )(),
    rate_limiter: RateLimiter | None = None,
    quota: QuotaStore | None = None,
    max_open_runs: int | None = None,
    max_streams: int = 8,
    max_stream_s: float = 900.0,
    keepalive_s: float = 15.0,
    turn_timeout_s: float = 300.0,
    max_body_bytes: int = 1 << 20,
    cors_origins: tuple[str, ...] = (),
    event_view: EventView = default_event_view,
    mcp: MCPToolServer | None = None,
    mcp_origins: tuple[str, ...] = (),
    a2a: A2AConfig | None = None,
)

create_app ¶

create_app(
    cfg: ServerConfig, *, manage_engine: bool = False
) -> Starlette

The ASGI app. With manage_engine=True the app starts and stops cfg.engine with the server (what aaa-server serve does); otherwise you own the engine's lifetime.

default_event_view ¶

default_event_view(e: Event) -> dict[str, Any] | None

The projection of the log a client may see. Anything not listed here is not sent: system prompts, memory contents, guard internals and raw model requests stay inside.

Principal `dataclass` ¶

Principal(
    tenant: str,
    subject: str,
    scopes: frozenset[str] = frozenset(
        {SCOPE_RUN, SCOPE_SESSION}
    ),
)

Authenticator ¶

Bases: Protocol

authenticate `async` ¶

authenticate(headers: Mapping[str, str]) -> Principal

Return the caller or raise AuthError. headers keys are lower-case.

ApiKeyAuthenticator ¶

ApiKeyAuthenticator(keys: Sequence[ApiKey])

Static API keys, held as hashes. Every candidate is compared in constant time, so the response time does not reveal how much of a key was right or which keys exist.

from_keys `classmethod` ¶

from_keys(
    mapping: Mapping[
        str,
        tuple[str, str] | tuple[str, str, Sequence[str]],
    ],
) -> ApiKeyAuthenticator

{"<the key>": ("tenant", "subject"[, scopes])} for tests and small deployments. Prefer ApiKey(key_hash=...) entries loaded from a secret store.

JWTAuthenticator `dataclass` ¶

JWTAuthenticator(
    issuer: str,
    audience: str,
    keys: JWKSKeys | None = None,
    secret: str | None = None,
    algorithms: tuple[str, ...] = (
        "RS256",
        "ES256",
        "EdDSA",
    ),
    tenant_claim: str = "tenant",
    scope_claim: str = "scope",
    leeway_s: float = 30.0,
    default_scopes: frozenset[str] = (
        lambda: frozenset({SCOPE_RUN, SCOPE_SESSION})
    )(),
)

Verifies a bearer JWT. Safe by construction:

the algorithm allow-list is fixed by YOU (asymmetric by default); the token's own alg header is never trusted, and none and the HS* family are refused unless you configure a shared secret (and then only HS*);
exp, iss and aud are required and checked;
the tenant, subject and scopes come from claims you name, and a token without a tenant claim is rejected.

JWKSKeys ¶

JWKSKeys(
    url: str,
    *,
    client: AsyncClient | None = None,
    ttl_s: float = 600.0,
    min_refetch_s: float = 30.0,
    clock: Callable[[], float] | None = None,
)

Signing keys fetched from a JWKS URL and cached; an unknown kid triggers one refetch (rotation), rate-limited so a stream of bogus tokens cannot make the server hammer the identity provider.

RateLimiter ¶

Bases: Protocol

InMemoryRateLimiter ¶

InMemoryRateLimiter(
    per_second: float,
    burst: float,
    *,
    clock: Callable[[], float] | None = None,
)

Token bucket per key: burst tokens, refilled at per_second. One process only; use the Redis limiter to share a limit between server processes.

RedisRateLimiter ¶

RedisRateLimiter(
    redis: Any,
    per_second: float,
    burst: float,
    *,
    prefix: str = "aaa:rl:",
)

The same token bucket in Redis, evaluated atomically by a Lua script, so every server process shares one limit. Time comes from the Redis server (TIME), not from the callers' clocks.

Quota ¶

Bases: BaseModel

Per-tenant limits per UTC day. None = unlimited. max_cost is checked against spend SETTLED so far (a run's cost is recorded when it finishes), so concurrent runs can overshoot it by what they are still spending; pair it with per-run budgets.

QuotaStore ¶

QuotaStore(
    store: EventStore,
    default: Quota,
    *,
    overrides: dict[str, Quota] | None = None,
    clock: Callable[[], datetime] | None = None,
)

admit `async` ¶

admit(tenant: str, unit_id: str) -> None

Reserve one run/turn against today's quota or raise Denied(429). Idempotent per unit_id.

settle `async` ¶

settle(
    tenant: str, unit_id: str, *, tokens: int, cost: float
) -> None

Record what a finished run actually used (idempotent per unit_id).

Denied ¶

Denied(
    status: int,
    code: str,
    message: str,
    retry_after: float | None = None,
)

Bases: Exception

A request was refused before any work started. status is the HTTP status to answer with.

A2AConfig `dataclass` ¶

A2AConfig(
    algorithm: str,
    url: str,
    name: str = "AAA agent",
    description: str = "An agent served by aaa-server.",
    version: str = "0.1",
    skills: tuple[dict[str, Any], ...] = (),
    to_input: Callable[
        [str], dict[str, Any]
    ] = lambda text: {"task": text},
    result_text: Callable[[Any], str] = _default_text,
    wait_s: float = 300.0,
    extra: dict[str, Any] = dict(),
)

url is this server's public address (the card must carry an absolute URL).

aaa_server¶

ServerConfig dataclass ¶