14. Client Libraries

Chapter 14 of 18 · 15 min

KEY INSIGHT

Client libraries abstract HTTP complexity into idiomatic interfacesΓÇöreliable error handling and automatic retry logic reduce integration failures for every consumer. Raw HTTP client code repeats authentication headers, base URL configuration, and error handling across endpoints. Client libraries encapsulate these patterns, providing clean methods that map to API operations. Consumers focus on business logic rather than connection management. Python clients typically use `httpx` for async support. Type stubs enable autocomplete for method signatures and parameters. Response parsing validates server payloads against expected schemas, catching mismatches before they propagate through business logic. ```python from httpx import AsyncClient, Response, TimeoutException from pydantic import BaseModel from typing import Optional import asyncio class InferenceError(Exception): def __init__(self, status_code: int, message: str): self.status_code = status_code self.message = message super().__init__(f"{status_code}: {message}") class CompletionRequest(BaseModel): model: str messages: list[dict] temperature: float = 0.7 class CompletionResponse(BaseModel): id: str model: str content: str usage: dict class AsyncInferenceClient: def __init__(self, base_url: str, api_key: str, timeout: float = 30.0): self.base_url = base_url.rstrip("/") self.api_key = api_key self._client = AsyncClient( timeout=timeout, headers={"Authorization": f"Bearer {api_key}"} ) async def completions(self, request: CompletionRequest) -> CompletionResponse: for attempt in range(3): try: response = await self._client.post( f"{self.base_url}/v1/chat/completions", json=request.model_dump() ) if response.status_code == 429: await asyncio.sleep(2 ** attempt) continue if response.status_code >= 400: raise InferenceError(response.status_code, response.text) return CompletionResponse(**response.json()) except TimeoutException: if attempt == 2: raise InferenceError(408, "Request timeout") await asyncio.sleep(1) raise InferenceError(503, "Service unavailable after retries") async def close(self): await self._client.aclose() # Usage async def main(): client = AsyncInferenceClient("http://localhost:8000", "test-key") try: result = await client.completions( CompletionRequest(model="llama3.2", messages=[{"role": "user", "content": "Hello"}]) ) print(result.content) finally: await client.close() ``` Retry logic handles transient failures. Exponential backoff with jitter prevents thundering herd problems when the API recovers. Timeout configuration prevents hanging indefinitely on unresponsive services. Include connection pooling configuration for high-throughput applications. Set keepalive timeouts to prevent stale connections. Add request ID headers for tracing across service boundaries.

EXERCISE

Extend the client library with streaming support using Server-Sent Events. Handle the SSE format parsing and provide an async generator yielding completion chunks as they arrive.