AIR

Structured events

AIR emits a structured Event object for every routing decision that matters: provider switches, health changes, exhaustion, recovery, usage thresholds, and capacity based skips. This works independently of any console output or TTS, so a web backend, bot, or headless service can consume it directly, for example by forwarding events over its own WebSocket or SSE connection, or into logs or analytics. AIR does not implement WebSockets, SSE, or any transport itself; you decide what to do with each Event.

The Event type

python
from air.events import Event

Event is a dataclass (air.events.Event) with these fields, all set only when relevant to that event type, otherwise None:

| field | type | meaning | |---|---|---| | type | str | one of the event type constants below | | message | str | human readable text, prefixed with [AIR]: | | provider | str \| None | the provider name this event is about | | model | str \| None | not currently populated by any built in event | | reason | str \| None | free text reason, meaning depends on event type | | previous_provider | str \| None | set on provider_switch events | | next_provider | str \| None | set on provider_switch events | | used_tokens | int \| None | set on usage_threshold events | | limit_tokens | int \| None | set on usage_threshold events | | threshold | float \| None | the configured threshold fraction that was crossed, set on usage_threshold events | | estimated_tokens | int \| None | set on capacity_skipped events | | remaining_tokens | int \| None | set on capacity_skipped events |

Event type constants

python
from air.events import (
    PROVIDER_SWITCH,       # "provider_switch"
    PROVIDER_EXHAUSTED,    # "provider_exhausted"
    PROVIDER_RECOVERED,    # "provider_recovered"
    USAGE_THRESHOLD,       # "usage_threshold"
    ALL_EXHAUSTED,         # "all_exhausted"
    CAPACITY_RESTORED,     # "capacity_restored"
    CAPACITY_SKIPPED,      # "capacity_skipped"
)

There is also a "health_change" event type emitted for health transitions that are not specifically exhausted (for example, a transition into rate_limited or temporarily_unavailable). It is not exposed as a named constant in air.events in the current source; match on the literal string "health_change" if you need to handle it separately from the six named constants above.

  • provider_switch: emitted when AIR falls back from one provider to the next after a fallback eligible failure. Carries previous_provider, next_provider, and reason (a short phrase such as "is rate limited").
  • provider_exhausted: emitted when a provider's health transitions into exhausted (a QuotaExceededError).
  • provider_recovered: emitted when a provider's health transitions from anything other than available back to available.
  • capacity_restored: emitted alongside provider_recovered, specifically when the previous status was exhausted.
  • usage_threshold: emitted the first time a provider's usage crosses a configured threshold (see capacity-tracking.md).
  • capacity_skipped: emitted when a provider is skipped before being called, either because it is still marked exhausted, or because its estimated remaining capacity is below the current request's estimated size.
  • all_exhausted: emitted once, right before ProvidersExhaustedError is raised, when every provider failed or was skipped.

Subscribing

python
def handler(event):
    print(event.type, event.provider, event.reason)
 
air.on_event(handler)

air.on_event(callback) registers a callback. Callbacks may be a normal function or an async def function; AIR awaits the result if it is awaitable. There is no filtering built in for on_event; every dispatched event goes to every registered callback, so filter by event.type inside your own callback if you only care about some event types.

python
air.off_event(handler)

air.off_event(callback) removes a previously registered callback (comparing by equality; passing the exact same function/object you registered works, a different function that merely looks the same does not).

Failure isolation

If a callback raises an exception (sync) or its awaited coroutine raises (async), AIR catches and discards it. A broken event callback never breaks routing, fallback, or the returned ChatResponse.

structured_events on ChatResponse

python
response = await air.chat("Hello")
for event in response.structured_events:
    print(event.type)

ChatResponse.structured_events is a list[Event] containing every event emitted during that specific chat(...) call (for example, a provider_switch followed eventually by whichever health event the winning provider produced). This lets you inspect what happened for a single call without registering a callback at all. ChatResponse.events (a list[str]) contains the same events' human readable message strings, for simple logging.

Note that when every provider fails and AIR raises ProvidersExhaustedError instead of returning a ChatResponse, there is no response to read structured_events from. The all_exhausted event (and everything before it in that call) is still delivered to on_event callbacks; use those if you need to observe a fully failed call.

Example: forwarding events without console or TTS

python
import asyncio
from air import AIR
from air.providers.groq import GroqProvider
 
events_for_websocket = []
 
def forward(event):
    events_for_websocket.append({"type": event.type, "provider": event.provider, "reason": event.reason})
 
async def main():
    air = AIR()
    air.on_event(forward)
    air.add_provider(GroqProvider(api_key="...", model="llama-3.3-70b-versatile"))
    await air.chat("Hello")
    # events_for_websocket now holds anything AIR emitted during that call
    # send it over your own WebSocket/SSE connection here
 
asyncio.run(main())

Nothing about this requires a terminal, an audio device, or TTS to be enabled.