Guidelines for writing correct, performant async Python code across VoxAgent's FastAPI API layer, provider implementations, and core pipeline modules.
-
async-no-blocking — NEVER call blocking I/O in async functions. Use aiohttp/httpx instead of requests, aiofiles instead of open(), asyncio.sleep() instead of time.sleep(). If you must use blocking code, wrap with asyncio.to_thread().
-
async-gather-parallel — Run independent async operations concurrently with asyncio.gather():
result_a = await fetch_a()
result_b = await fetch_b()
result_a, result_b = await asyncio.gather(fetch_a(), fetch_b())
-
async-timeout — Always set timeouts on external calls. Use asyncio.wait_for() or httpx timeout parameter:
async with httpx.AsyncClient(timeout=10.0) as client:
response = await client.get(url)
-
async-context-manager — Use async with for resources that need cleanup (HTTP clients, DB connections, file handles):
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
data = await response.json()
-
async-generator — Use async for with async generators for streaming responses:
async def stream_response(provider: LLMProvider) -> AsyncGenerator[str, None]:
async for chunk in provider.chat_stream(messages):
yield chunk
-
fastapi-dependency-injection — Use Depends() for shared resources. Never instantiate providers or DB connections inside endpoint functions:
async def get_provider(name: str = Query(...)) -> LLMProvider:
return registry.get_llm(name)
@router.post("/chat")
async def chat(provider: LLMProvider = Depends(get_provider)):
...
-
fastapi-pydantic-models — All request/response bodies MUST use Pydantic BaseModel. Never accept raw dicts:
class ChatRequest(BaseModel):
messages: list[Message]
model: str = "default"
class ChatResponse(BaseModel):
response: str
model_used: str
latency_ms: float
-
fastapi-lifespan — Use lifespan context manager for startup/shutdown, not @app.on_event:
@asynccontextmanager
async def lifespan(app: FastAPI):
await registry.initialize()
yield
await registry.shutdown()
app = FastAPI(lifespan=lifespan)
-
fastapi-background-tasks — Use BackgroundTasks for non-critical post-response work (logging, analytics). Never use for critical operations:
@router.post("/execute")
async def execute(background_tasks: BackgroundTasks):
result = await skill.execute(intent)
background_tasks.add_task(log_execution, result)
return result
-
fastapi-streaming — Use StreamingResponse with async generators for real-time output:
@router.post("/chat/stream")
async def chat_stream(request: ChatRequest):
return StreamingResponse(
provider.chat_stream(request.messages),
media_type="text/event-stream",
)
-
fastapi-exception-handlers — Register custom exception handlers. Map domain exceptions to HTTP status codes:
@app.exception_handler(ProviderUnavailableError)
async def provider_error_handler(request, exc):
return JSONResponse(status_code=503, content={"detail": str(exc)})
-
provider-abstract-base — Every provider MUST implement all abstract methods from its base class in providers/base.py. No partial implementations.
-
provider-health-check — Every provider MUST implement a health check method that verifies connectivity without side effects:
async def health_check(self) -> bool:
try:
await self._client.get("/models", timeout=5.0)
return True
except Exception:
return False
-
provider-retry — Use exponential backoff for transient failures. Max 3 retries, starting at 1s:
async def _request_with_retry(self, fn, max_retries: int = 3) -> Any:
for attempt in range(max_retries):
try:
return await fn()
except (httpx.TimeoutException, httpx.HTTPStatusError) as e:
if attempt == max_retries - 1:
raise
await asyncio.sleep(2 ** attempt)
-
provider-fallback-chain — When primary provider fails, fall back through the chain defined in config. Never silently switch providers — log the fallback:
logger.warning("Provider %s failed, falling back to %s", primary.name, fallback.name)
-
provider-session-reuse — Create httpx.AsyncClient once in __init__ or lifespan, reuse across requests. Never create a new client per request:
class OpenAIProvider(LLMProvider):
def __init__(self, config: ProviderConfig):
self._client = httpx.AsyncClient(
base_url="https://api.openai.com/v1",
headers={"Authorization": f"Bearer {config.api_key}"},
timeout=30.0,
)
-
error-async-context — Always include async context in error logs (which provider, which operation, which request):
logger.error("Chat failed: provider=%s model=%s error=%s", self.name, model, str(e))
-
error-structured-logging — Use structured logging with logging module. Include request_id, provider, operation:
logger.info("Request completed", extra={"request_id": req_id, "latency_ms": latency})
-
error-graceful-degradation — API endpoints should return partial results when possible, not fail entirely:
providers = await asyncio.gather(*health_checks, return_exceptions=True)
healthy = [p for p in providers if not isinstance(p, Exception)]
-
error-timeout-separate — Distinguish timeout errors from other errors. Timeouts often need different recovery:
except asyncio.TimeoutError:
logger.warning("Provider %s timed out after %ds", name, timeout)
except httpx.HTTPStatusError as e:
logger.error("Provider %s returned %d", name, e.response.status_code)
-
test-pytest-asyncio — All async tests use @pytest.mark.asyncio (or asyncio_mode = "auto" in config). Never use asyncio.run() in tests:
async def test_provider_chat():
provider = MockLLMProvider()
result = await provider.chat(messages)
assert result == expected
-
test-mock-async — Use AsyncMock for mocking async methods:
from unittest.mock import AsyncMock
provider = AsyncMock(spec=LLMProvider)
provider.chat.return_value = "mocked response"
-
test-httpx-mock — Use respx or pytest-httpx for mocking HTTP calls. Never make real network calls:
@pytest.fixture
def mock_api(respx_mock):
respx_mock.post("https://api.openai.com/v1/chat/completions").respond(
json={"choices": [{"message": {"content": "test"}}]}
)
-
test-event-loop — Never create your own event loop in tests. Let pytest-asyncio manage it. If testing callbacks, use asyncio.Event:
async def test_callback():
received = asyncio.Event()
async def on_result(data):
received.set()
await process(callback=on_result)
await asyncio.wait_for(received.wait(), timeout=5.0)
