Example: structured events

import asyncio
 
from air import AIR, ChatRequest, ChatResponse, Provider
from air.exceptions import RateLimitError
 
received_events = []
 
 
def forward_event(event):
    # In a real backend, send this over your own WebSocket/SSE connection,
    # write it to a log, or feed it to analytics. No console or TTS involved.
    received_events.append({"type": event.type, "provider": event.provider, "reason": event.reason})
 
 
class FlakyProvider(Provider):
    name = "flaky"
 
    async def chat(self, request: ChatRequest) -> ChatResponse:
        raise RateLimitError("simulated rate limit", provider=self.name)
 
 
class BackupProvider(Provider):
    name = "backup"
 
    async def chat(self, request: ChatRequest) -> ChatResponse:
        return ChatResponse(content="ok", provider=self.name, model="demo")
 
 
async def main():
    air = AIR()
    air.on_event(forward_event)
    air.add_provider(FlakyProvider(), priority=1)
    air.add_provider(BackupProvider(), priority=2)
 
    response = await air.chat("Hello")
 
    print("via callback:", received_events)
    print("via response.structured_events:", [e.type for e in response.structured_events])
 
 
if __name__ == "__main__":
    asyncio.run(main())