HinterBuild logoHinterBuild
Back to blog
AI Systems · 21 min read

Async LLM API Calls with FastAPI and Server-Sent Events

Build async LLM API calls with FastAPI: SSE streaming, semaphore concurrency control, retries, circuit breakers and deployment patterns for production.

Muhammad Abdul Sami, author

Muhammad Abdul Sami

· 21 min read

  • FastAPI
  • Streaming
  • Python
  • APIs
  • LLM
  • Performance

Building production LLM APIs requires handling concurrent requests efficiently while streaming responses to users in real time. Async LLM API calls with FastAPI and Server-Sent Events (SSE) are the standard way to do that: the event loop waits on the provider without tying up a worker, and SSE delivers tokens to the browser over plain HTTP. This guide covers the async patterns, the SSE implementation, concurrency control, and the production hardening we apply to systems processing 10,000+ AI requests daily.

Key Takeaways:

  • An LLM call spends almost all of its wall-clock time waiting on the provider, so a sync FastAPI endpoint wastes a worker per request; async def plus an async client lets one worker hold hundreds of in-flight calls.
  • Use StreamingResponse with text/event-stream and the data: ...\n\n framing; the browser EventSource API only supports GET, so POST-based chat streams need fetch plus a ReadableStream reader on the client.
  • Bound concurrency with an asyncio.Semaphore inside the generator, not around the return StreamingResponse(...); otherwise the permit is released before the first token is sent.
  • Retry only transient errors (429, timeouts, connection resets) with exponential backoff and jitter; fail fast on 4xx auth and validation errors.
  • Wrap the provider in a circuit breaker so an outage returns 503 in milliseconds instead of burning your timeout budget on every request.
  • Use idle timeouts (no token for N seconds) for streams and total timeouts for non-streaming calls; they fail in different ways.

Table of Contents:

Why Async Matters for LLM APIs

LLM APIs are inherently I/O-bound - you spend 90% of the time waiting for the LLM provider to generate tokens. Synchronous implementations waste server resources and limit throughput dramatically.

Illustrative comparison on a single 4-worker node (numbers are typical of what we see; your provider latency and token counts will move them):

PatternRequests/secMemory UsageLatency P95
Sync blocking12 req/s450MB8.5s
Async/await180 req/s280MB1.2s
Async + streaming240 req/s180MB0.8s

The async pattern increases throughput by an order of magnitude while cutting memory, because each in-flight request costs a coroutine frame instead of a thread stack. The mechanism is simple: FastAPI runs async def path operations directly on the event loop, so an await client.chat.completions.create(...) yields control while the socket is idle. A def endpoint, by contrast, is pushed to a threadpool with a default cap of 40 threads, which is the real ceiling most "slow FastAPI" reports are hitting.

When to Use Async LLM APIs

Use async when:

  • Handling multiple concurrent requests
  • Streaming responses to users
  • Integrating with other async services (databases, caches)
  • Building AI agent systems with tool calling
  • Processing RAG pipelines with retrieval

Stick with sync when:

  • Single-user scripts or notebooks
  • Batch processing without concurrency
  • Simple prototypes without streaming

For production backend API engineering, async is non-negotiable.

FastAPI Streaming Architecture

FastAPI's async support makes it ideal for LLM APIs. Here's the production architecture:

python
from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field
from openai import AsyncOpenAI
import asyncio
from typing import AsyncGenerator

app = FastAPI()
client = AsyncOpenAI()

class ChatRequest(BaseModel):
    messages: list[dict]
    model: str = "gpt-4"
    temperature: float = Field(default=0.7, ge=0, le=2)
    max_tokens: int = Field(default=2000, le=4000)
    stream: bool = True

class ChatResponse(BaseModel):
    content: str
    usage: dict
    model: str

@app.post("/v1/chat/completions")
async def chat_completions(request: ChatRequest):
    """Async LLM endpoint with streaming support."""
    
    if request.stream:
        return StreamingResponse(
            stream_llm_response(request),
            media_type="text/event-stream"
        )
    else:
        response = await get_complete_response(request)
        return response

async def stream_llm_response(
    request: ChatRequest
) -> AsyncGenerator[str, None]:
    """Stream LLM tokens as Server-Sent Events."""
    
    try:
        stream = await client.chat.completions.create(
            model=request.model,
            messages=request.messages,
            temperature=request.temperature,
            max_tokens=request.max_tokens,
            stream=True
        )
        
        async for chunk in stream:
            if chunk.choices[0].delta.content:
                content = chunk.choices[0].delta.content
                yield f"data: {json.dumps({'content': content})}\n\n"
                
    except Exception as e:
        error_msg = f"data: {json.dumps({'error': str(e)})}\n\n"
        yield error_msg
    finally:
        yield "data: [DONE]\n\n"

async def get_complete_response(request: ChatRequest) -> ChatResponse:
    """Non-streaming async completion."""
    
    response = await client.chat.completions.create(
        model=request.model,
        messages=request.messages,
        temperature=request.temperature,
        max_tokens=request.max_tokens,
        stream=False
    )
    
    return ChatResponse(
        content=response.choices[0].message.content,
        usage=response.usage.dict(),
        model=response.model
    )

Key patterns:

  • AsyncGenerator for streaming responses
  • Pydantic validation on requests
  • Separate streaming and non-streaming paths
  • Proper SSE format with data: prefix

Server-Sent Events Implementation

Server-Sent Events (SSE) provide one-way real-time communication from server to client. Unlike WebSockets, SSE works over HTTP and reconnects automatically.

SSE Format Requirements

python
import json
from typing import AsyncGenerator

async def format_sse_message(
    event_type: str,
    data: dict
) -> str:
    """Format data as SSE message."""
    
    # SSE format: "event: <type>\ndata: <json>\n\n"
    return f"event: {event_type}\ndata: {json.dumps(data)}\n\n"

async def stream_with_events(
    messages: list[dict]
) -> AsyncGenerator[str, None]:
    """Stream with typed events."""
    
    # Initial event
    yield await format_sse_message("start", {"status": "streaming"})
    
    try:
        stream = await client.chat.completions.create(
            model="gpt-4",
            messages=messages,
            stream=True
        )
        
        token_count = 0
        
        async for chunk in stream:
            if chunk.choices[0].delta.content:
                token_count += 1
                
                # Token event
                yield await format_sse_message(
                    "token",
                    {
                        "content": chunk.choices[0].delta.content,
                        "index": token_count
                    }
                )
                
        # Completion event
        yield await format_sse_message(
            "done",
            {
                "tokens": token_count,
                "status": "complete"
            }
        )
        
    except Exception as e:
        # Error event
        yield await format_sse_message(
            "error",
            {
                "message": str(e),
                "type": type(e).__name__
            }
        )

Client-Side SSE Consumption

javascript
// Frontend SSE consumer
const eventSource = new EventSource('/v1/chat/completions');

eventSource.addEventListener('start', (e) => {
    const data = JSON.parse(e.data);
    console.log('Stream started:', data.status);
});

eventSource.addEventListener('token', (e) => {
    const data = JSON.parse(e.data);
    appendToUI(data.content);
});

eventSource.addEventListener('done', (e) => {
    const data = JSON.parse(e.data);
    console.log(`Complete: ${data.tokens} tokens`);
    eventSource.close();
});

eventSource.addEventListener('error', (e) => {
    const data = JSON.parse(e.data);
    console.error('Stream error:', data.message);
    eventSource.close();
});

SSE advantages for LLM APIs:

  • Automatic reconnection with Last-Event-ID
  • Simpler than WebSockets for one-way streams
  • Works through proxies and firewalls
  • Native browser support with EventSource

One caveat with the client above: the EventSource API defined in the WHATWG HTML spec only issues GET requests and cannot send a body or custom headers. A chat endpoint that takes a POST body with messages needs a different consumer: call fetch() with the JSON body, read response.body.getReader(), decode chunks, and split on the blank-line frame delimiter yourself. That is exactly how the OpenAI and Anthropic browser SDKs consume their streaming endpoints; the OpenAI streaming docs and Anthropic streaming docs document the same data: framing this guide produces. EventSource is still the right choice for GET-style feeds such as job progress.

Two server-side details that bite in production:

  • Buffering proxies. Nginx buffers responses by default and will hold your tokens until the response completes. Send X-Accel-Buffering: no and Cache-Control: no-cache headers on the streaming response, and disable proxy_buffering for the route.
  • Keep-alives. Idle connections through load balancers get cut after 30-60 seconds. Emit an SSE comment line (: ping\n\n) every 15 seconds while waiting on a slow first token; the browser ignores comments but the connection stays warm.

Choosing a Streaming Transport

SSE is the default for token streaming, but it is not the only option. The table summarises when each transport fits an LLM API.

TransportDirectionWorks through HTTP/1.1 proxiesAuto reconnectRequest bodyBest for
SSE (text/event-stream)Server to clientYesYes (EventSource)GET only via EventSource; POST via fetchToken streaming, progress feeds
Fetch + ReadableStream (NDJSON)Server to clientYesManualYesSame as SSE when you control the client and want custom framing
WebSocketBidirectionalUsually, with upgrade supportManualN/A (messages)Voice, collaborative editing, interrupting generation mid-stream
Long pollingClient pullYesN/AYesLegacy clients, very low update rates

Pick WebSockets only when the client needs to talk back during generation (barge-in, live cancellation with state). For everything else SSE is cheaper to operate. Our WebSockets vs SSE vs long polling guide goes deeper on the trade-offs.

Concurrent Request Handling

Production LLM APIs must handle hundreds of concurrent requests without exhausting resources or hitting rate limits.

Semaphore-Based Concurrency Control

python
import asyncio
from contextlib import asynccontextmanager

class LLMRateLimiter:
    """Control concurrent LLM requests with semaphore."""
    
    def __init__(self, max_concurrent: int = 50):
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.active_requests = 0
        self.total_requests = 0
    
    @asynccontextmanager
    async def acquire(self):
        """Acquire semaphore with metrics."""
        
        async with self.semaphore:
            self.active_requests += 1
            self.total_requests += 1
            
            try:
                yield
            finally:
                self.active_requests -= 1
    
    def get_metrics(self) -> dict:
        """Return current metrics."""
        return {
            "active": self.active_requests,
            "total": self.total_requests,
            "capacity": self.semaphore._value
        }

# Global limiter instance
llm_limiter = LLMRateLimiter(max_concurrent=50)

@app.post("/v1/chat/completions")
async def chat_completions_with_limit(request: ChatRequest):
    """Chat endpoint with concurrency control."""
    
    async with llm_limiter.acquire():
        if request.stream:
            return StreamingResponse(
                stream_llm_response(request),
                media_type="text/event-stream"
            )
        else:
            return await get_complete_response(request)

@app.get("/metrics")
async def get_metrics():
    """Expose limiter metrics."""
    return llm_limiter.get_metrics()

Backpressure Handling

python
from fastapi import BackgroundTasks, status
from datetime import datetime, timedelta

class RequestQueue:
    """Queue system with backpressure."""
    
    def __init__(self, max_queue_size: int = 200):
        self.queue = asyncio.Queue(maxsize=max_queue_size)
        self.processing = False
    
    async def enqueue(
        self,
        request: ChatRequest,
        timeout: float = 30.0
    ) -> ChatResponse:
        """Enqueue request with timeout."""
        
        try:
            # Try to add to queue
            await asyncio.wait_for(
                self.queue.put(request),
                timeout=5.0  # Max 5s wait to queue
            )
        except asyncio.TimeoutError:
            raise HTTPException(
                status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
                detail="Queue full, try again later"
            )
        
        # Wait for processing
        try:
            response = await asyncio.wait_for(
                self._process_from_queue(),
                timeout=timeout
            )
            return response
        except asyncio.TimeoutError:
            raise HTTPException(
                status_code=status.HTTP_504_GATEWAY_TIMEOUT,
                detail="Request processing timeout"
            )
    
    async def _process_from_queue(self) -> ChatResponse:
        """Process next item from queue."""
        request = await self.queue.get()
        
        try:
            return await get_complete_response(request)
        finally:
            self.queue.task_done()

# Global queue
request_queue = RequestQueue(max_queue_size=200)

@app.post("/v1/chat/queue")
async def queued_completions(request: ChatRequest):
    """Queue-based endpoint for backpressure handling."""
    return await request_queue.enqueue(request)

Concurrency best practices:

  • Use semaphores to limit concurrent LLM calls
  • Implement queue systems for backpressure
  • Return 503 when overloaded (don't queue indefinitely)
  • Monitor active request counts
  • Set per-user rate limits

See our rate limiting guide for more patterns.

Async LLM API Pitfalls

These are the mistakes we find most often when reviewing async LLM APIs. Each one produces a system that looks correct in a demo and degrades under load.

Releasing the Semaphore Before the Stream Starts

Look closely at chat_completions_with_limit above. The async with llm_limiter.acquire() block returns a StreamingResponse object; the generator inside it has not run yet. The context manager exits, the permit is released, and only then does Starlette start iterating the generator and calling the provider. The semaphore limits how fast you can create responses, not how many streams are in flight.

The fix is to acquire the permit inside the generator so it is held for the lifetime of the stream:

python
async def stream_llm_response_limited(request: ChatRequest):
    async with llm_limiter.acquire():
        async for chunk in stream_llm_response(request):
            yield chunk

The same reasoning applies to database sessions, tracing spans, and anything else you open around a streaming return.

Blocking the Event Loop

One synchronous call inside an async def endpoint (a sync HTTP client, time.sleep, a CPU-heavy tokenizer call, a blocking DB driver) stalls every other coroutine on that worker. Symptoms are latency spikes that correlate with traffic rather than with provider latency. Audit with PYTHONASYNCIODEBUG=1 in staging, which logs callbacks that take longer than 100 ms, and push unavoidable blocking work through asyncio.to_thread or run_in_executor.

Unbounded asyncio.gather

The batch endpoint above fans out up to 50 provider calls at once. Behind a semaphore that is fine; without one, a burst of batch requests multiplies your concurrency by 50 and trips the provider's rate limit for every other tenant on the same key. Always route fan-out through the same limiter as single requests.

Client Disconnects That Keep Generating

When a user closes the tab, the generator keeps pulling tokens from the provider and you keep paying for them. Check await request.is_disconnected() between chunks (pass the Request into the generator) and cancel the upstream stream when it returns true. The sse-starlette package handles this and the keep-alive pings for you if you prefer not to hand-roll it.

Retrying Non-Idempotent Streams

Retrying a stream after the client has already received half of the tokens produces a duplicated or contradictory answer. Retry only before the first token has been sent; after that, emit an error event and let the client decide whether to re-request.

Production Patterns

Real production systems need more than basic async - they require retry logic, circuit breakers, and proper error handling.

Retry with Exponential Backoff

python
import asyncio
from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential,
    retry_if_exception_type
)
from openai import RateLimitError, APITimeoutError

@retry(
    retry=retry_if_exception_type((RateLimitError, APITimeoutError)),
    wait=wait_exponential(multiplier=1, min=2, max=30),
    stop=stop_after_attempt(5)
)
async def llm_call_with_retry(
    messages: list[dict],
    model: str = "gpt-4"
) -> str:
    """LLM call with automatic retry."""
    
    response = await client.chat.completions.create(
        model=model,
        messages=messages,
        timeout=30.0
    )
    
    return response.choices[0].message.content

Circuit Breaker Pattern

python
from datetime import datetime, timedelta
from enum import Enum

class CircuitState(Enum):
    CLOSED = "closed"  # Normal operation
    OPEN = "open"      # Blocking requests
    HALF_OPEN = "half_open"  # Testing recovery

class CircuitBreaker:
    """Circuit breaker for LLM API calls."""
    
    def __init__(
        self,
        failure_threshold: int = 5,
        timeout: int = 60,
        expected_exception=Exception
    ):
        self.failure_threshold = failure_threshold
        self.timeout = timeout
        self.expected_exception = expected_exception
        
        self.failure_count = 0
        self.last_failure_time = None
        self.state = CircuitState.CLOSED
    
    async def call(self, func, *args, **kwargs):
        """Execute function with circuit breaker."""
        
        if self.state == CircuitState.OPEN:
            if self._should_attempt_reset():
                self.state = CircuitState.HALF_OPEN
            else:
                raise Exception("Circuit breaker is OPEN")
        
        try:
            result = await func(*args, **kwargs)
            self._on_success()
            return result
            
        except self.expected_exception as e:
            self._on_failure()
            raise e
    
    def _on_success(self):
        """Reset on successful call."""
        self.failure_count = 0
        self.state = CircuitState.CLOSED
    
    def _on_failure(self):
        """Increment failure count."""
        self.failure_count += 1
        self.last_failure_time = datetime.now()
        
        if self.failure_count >= self.failure_threshold:
            self.state = CircuitState.OPEN
    
    def _should_attempt_reset(self) -> bool:
        """Check if enough time passed to retry."""
        return (
            self.last_failure_time and
            datetime.now() - self.last_failure_time >= 
            timedelta(seconds=self.timeout)
        )

# Global circuit breaker
llm_circuit_breaker = CircuitBreaker(
    failure_threshold=5,
    timeout=60
)

async def protected_llm_call(messages: list[dict]) -> str:
    """LLM call protected by circuit breaker."""
    
    return await llm_circuit_breaker.call(
        llm_call_with_retry,
        messages=messages
    )

Request Deduplication

python
import hashlib
from typing import Optional

class ResponseCache:
    """Deduplicate identical in-flight requests."""
    
    def __init__(self):
        self.pending: dict[str, asyncio.Future] = {}
    
    def _hash_request(self, request: ChatRequest) -> str:
        """Create hash of request."""
        key = f"{request.model}:{json.dumps(request.messages)}"
        return hashlib.sha256(key.encode()).hexdigest()
    
    async def get_or_create(
        self,
        request: ChatRequest
    ) -> ChatResponse:
        """Get cached response or create new request."""
        
        request_hash = self._hash_request(request)
        
        # Check if request already in flight
        if request_hash in self.pending:
            # Wait for existing request
            return await self.pending[request_hash]
        
        # Create new future
        future = asyncio.Future()
        self.pending[request_hash] = future
        
        try:
            # Execute request
            response = await get_complete_response(request)
            future.set_result(response)
            return response
            
        except Exception as e:
            future.set_exception(e)
            raise
        finally:
            # Remove from pending
            del self.pending[request_hash]

cache = ResponseCache()

@app.post("/v1/chat/dedup")
async def deduplicated_completions(request: ChatRequest):
    """Deduplicate identical concurrent requests."""
    return await cache.get_or_create(request)

These patterns prevent cascading failures and reduce costs. The retry decorator comes from tenacity; add wait_random_exponential rather than plain exponential backoff so that a fleet of workers does not retry in lockstep after a 429. Our event-driven architecture guide covers more production patterns.

Error Handling & Retries

Production LLM APIs must handle errors gracefully - rate limits, timeouts, and provider outages are inevitable.

Comprehensive Error Handler

python
from fastapi import Request
from fastapi.responses import JSONResponse
from openai import (
    RateLimitError,
    APITimeoutError,
    APIConnectionError,
    AuthenticationError
)
import logging

logger = logging.getLogger(__name__)

@app.exception_handler(RateLimitError)
async def rate_limit_handler(request: Request, exc: RateLimitError):
    """Handle OpenAI rate limit errors."""
    
    logger.warning(f"Rate limit hit: {exc}")
    
    return JSONResponse(
        status_code=429,
        content={
            "error": "rate_limit_exceeded",
            "message": "Too many requests, please retry after delay",
            "retry_after": 60
        },
        headers={"Retry-After": "60"}
    )

@app.exception_handler(APITimeoutError)
async def timeout_handler(request: Request, exc: APITimeoutError):
    """Handle timeout errors."""
    
    logger.error(f"API timeout: {exc}")
    
    return JSONResponse(
        status_code=504,
        content={
            "error": "gateway_timeout",
            "message": "LLM provider timeout, please retry"
        }
    )

@app.exception_handler(APIConnectionError)
async def connection_handler(request: Request, exc: APIConnectionError):
    """Handle connection errors."""
    
    logger.error(f"Connection error: {exc}")
    
    return JSONResponse(
        status_code=503,
        content={
            "error": "service_unavailable",
            "message": "Cannot reach LLM provider, please retry"
        }
    )

@app.exception_handler(AuthenticationError)
async def auth_handler(request: Request, exc: AuthenticationError):
    """Handle authentication errors."""
    
    logger.critical(f"Auth error: {exc}")
    
    return JSONResponse(
        status_code=500,
        content={
            "error": "internal_error",
            "message": "Configuration error, contact support"
        }
    )

Graceful Degradation

python
from typing import Optional

class LLMProvider(Enum):
    OPENAI = "openai"
    ANTHROPIC = "anthropic"
    FALLBACK = "fallback"

async def llm_with_fallback(
    messages: list[dict],
    primary: LLMProvider = LLMProvider.OPENAI
) -> tuple[str, LLMProvider]:
    """Try primary provider, fallback on failure."""
    
    providers = [
        (LLMProvider.OPENAI, call_openai),
        (LLMProvider.ANTHROPIC, call_anthropic),
    ]
    
    # Try primary first
    for provider, func in providers:
        if provider == primary:
            try:
                result = await func(messages)
                return result, provider
            except Exception as e:
                logger.warning(f"{provider} failed: {e}")
                continue
    
    # Try remaining providers
    for provider, func in providers:
        if provider != primary:
            try:
                result = await func(messages)
                logger.info(f"Fallback to {provider} succeeded")
                return result, provider
            except Exception as e:
                logger.error(f"{provider} fallback failed: {e}")
                continue
    
    raise Exception("All LLM providers failed")

async def call_openai(messages: list[dict]) -> str:
    """Call OpenAI API."""
    response = await openai_client.chat.completions.create(
        model="gpt-4",
        messages=messages
    )
    return response.choices[0].message.content

async def call_anthropic(messages: list[dict]) -> str:
    """Call Anthropic API."""
    response = await anthropic_client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=2000,
        messages=messages
    )
    return response.content[0].text

Error handling principles:

  • Retry transient errors (rate limits, timeouts)
  • Fail fast on auth/config errors
  • Provide fallback providers
  • Log all errors with context
  • Return actionable error messages to clients

Performance Optimization

Async alone isn't enough - you need connection pooling, caching, and smart batching.

Connection Pooling

The OpenAI and Anthropic SDKs are built on httpx, so you can hand them a tuned AsyncClient. Without limits, a burst of requests opens a new TLS connection per call and the handshake cost shows up as first-token latency.

python
from openai import AsyncOpenAI
import httpx

# Configure with connection limits
http_client = httpx.AsyncClient(
    limits=httpx.Limits(
        max_connections=100,
        max_keepalive_connections=20,
        keepalive_expiry=30.0
    ),
    timeout=httpx.Timeout(30.0, connect=5.0)
)

client = AsyncOpenAI(
    http_client=http_client,
    max_retries=3
)

Response Caching

python
from functools import lru_cache
import hashlib
import json

class LLMCache:
    """Cache LLM responses with TTL."""
    
    def __init__(self, ttl: int = 3600):
        self.cache: dict[str, tuple[ChatResponse, float]] = {}
        self.ttl = ttl
    
    def _key(self, request: ChatRequest) -> str:
        """Generate cache key."""
        data = json.dumps({
            "model": request.model,
            "messages": request.messages,
            "temperature": request.temperature
        }, sort_keys=True)
        return hashlib.sha256(data.encode()).hexdigest()
    
    def get(self, request: ChatRequest) -> Optional[ChatResponse]:
        """Get cached response if valid."""
        key = self._key(request)
        
        if key in self.cache:
            response, timestamp = self.cache[key]
            
            # Check TTL
            if time.time() - timestamp < self.ttl:
                return response
            else:
                del self.cache[key]
        
        return None
    
    def set(self, request: ChatRequest, response: ChatResponse):
        """Cache response."""
        key = self._key(request)
        self.cache[key] = (response, time.time())
    
    def clear_expired(self):
        """Remove expired entries."""
        now = time.time()
        expired = [
            k for k, (_, ts) in self.cache.items()
            if now - ts >= self.ttl
        ]
        for key in expired:
            del self.cache[key]

cache = LLMCache(ttl=3600)

@app.post("/v1/chat/cached")
async def cached_completions(request: ChatRequest):
    """Chat endpoint with caching."""
    
    # Check cache
    if cached := cache.get(request):
        return cached
    
    # Generate response
    response = await get_complete_response(request)
    
    # Cache for future requests
    cache.set(request, response)
    
    return response

Batch Request Processing

python
async def process_batch(
    requests: list[ChatRequest]
) -> list[ChatResponse]:
    """Process multiple requests concurrently."""
    
    tasks = [
        get_complete_response(req)
        for req in requests
    ]
    
    # Run all concurrently
    responses = await asyncio.gather(*tasks, return_exceptions=True)
    
    # Handle individual failures
    results = []
    for i, response in enumerate(responses):
        if isinstance(response, Exception):
            logger.error(f"Request {i} failed: {response}")
            results.append(None)
        else:
            results.append(response)
    
    return results

@app.post("/v1/chat/batch")
async def batch_completions(
    requests: list[ChatRequest]
) -> list[Optional[ChatResponse]]:
    """Batch processing endpoint."""
    
    if len(requests) > 50:
        raise HTTPException(
            status_code=400,
            detail="Maximum 50 requests per batch"
        )
    
    return await process_batch(requests)

The in-process cache above is a starting point; it is per-worker and does not survive restarts. For anything beyond one node, move it to Redis and consider semantic caching, which also catches paraphrased prompts. Check our LLM inference optimization guide if you are serving models yourself rather than calling a provider.

Deployment Checklist

Production async LLM APIs require proper configuration:

Environment Configuration

python
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    # API keys
    openai_api_key: str
    anthropic_api_key: Optional[str] = None
    
    # Concurrency
    max_concurrent_requests: int = 50
    max_queue_size: int = 200
    
    # Timeouts
    llm_timeout: float = 30.0
    request_timeout: float = 60.0
    
    # Circuit breaker
    circuit_breaker_threshold: int = 5
    circuit_breaker_timeout: int = 60
    
    # Caching
    cache_ttl: int = 3600
    cache_enabled: bool = True
    
    # Rate limiting
    rate_limit_requests: int = 100
    rate_limit_window: int = 60
    
    class Config:
        env_file = ".env"

settings = Settings()

Docker Deployment

dockerfile
FROM python:3.11-slim

WORKDIR /app

# Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy application
COPY . .

# Run with Gunicorn + Uvicorn workers
CMD ["gunicorn", "main:app", \
     "--workers", "4", \
     "--worker-class", "uvicorn.workers.UvicornWorker", \
     "--bind", "0.0.0.0:8000", \
     "--timeout", "120", \
     "--graceful-timeout", "30"]

Kubernetes Configuration

yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: llm-api
spec:
  replicas: 3
  selector:
    matchLabels:
      app: llm-api
  template:
    metadata:
      labels:
        app: llm-api
    spec:
      containers:
      - name: api
        image: llm-api:latest
        ports:
        - containerPort: 8000
        env:
        - name: MAX_CONCURRENT_REQUESTS
          value: "50"
        - name: LLM_TIMEOUT
          value: "30.0"
        resources:
          requests:
            memory: "512Mi"
            cpu: "500m"
          limits:
            memory: "1Gi"
            cpu: "1000m"
        livenessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 10
          periodSeconds: 30
        readinessProbe:
          httpGet:
            path: /ready
            port: 8000
          initialDelaySeconds: 5
          periodSeconds: 10
---
apiVersion: v1
kind: Service
metadata:
  name: llm-api
spec:
  selector:
    app: llm-api
  ports:
  - protocol: TCP
    port: 80
    targetPort: 8000
  type: LoadBalancer

Monitoring

python
from prometheus_client import Counter, Histogram, Gauge

# Metrics
llm_requests_total = Counter(
    'llm_requests_total',
    'Total LLM requests',
    ['model', 'status']
)

llm_request_duration = Histogram(
    'llm_request_duration_seconds',
    'LLM request duration',
    ['model']
)

llm_active_requests = Gauge(
    'llm_active_requests',
    'Currently active LLM requests'
)

@app.middleware("http")
async def metrics_middleware(request: Request, call_next):
    """Track request metrics."""
    
    llm_active_requests.inc()
    
    with llm_request_duration.labels(model="gpt-4").time():
        try:
            response = await call_next(request)
            llm_requests_total.labels(
                model="gpt-4",
                status=response.status_code
            ).inc()
            return response
        finally:
            llm_active_requests.dec()

Need help with deployment? Our backend API engineering team specializes in production AI infrastructure.

Frequently Asked Questions

What's the difference between async and streaming in LLM APIs?

Async refers to non-blocking I/O that allows handling multiple requests concurrently without threads. Streaming is about sending response tokens incrementally as they're generated, rather than waiting for completion. You can have async without streaming (concurrent non-streaming requests) or streaming without async (single streaming request blocking a thread). Production systems use both.

How many concurrent LLM requests can FastAPI handle?

With proper async patterns, a single FastAPI server (4 Uvicorn workers) can handle 200-300 concurrent LLM requests while maintaining sub-second response times. The bottleneck is usually LLM provider rate limits, not your server. Use semaphores to control concurrency and prevent overwhelming the provider.

Should I use WebSockets or Server-Sent Events for LLM streaming?

Use SSE for LLM streaming - it's simpler, works over HTTP, reconnects automatically, and is the standard for LLM APIs (OpenAI, Anthropic, etc.). WebSockets add complexity without benefits for one-way LLM → client communication. Reserve WebSockets for bidirectional real-time features like collaborative editing.

How do I handle OpenAI rate limits in production?

Implement three layers: (1) Semaphore to limit concurrent requests below your rate limit, (2) Exponential backoff retry for 429 errors, (3) Circuit breaker to stop sending requests during extended outages. Monitor your rate limit headers and adjust concurrency dynamically. Our rate limiting guide covers this in depth.

What's the ideal timeout for LLM API calls?

30 seconds for individual API calls with 3 retries is standard. For user-facing endpoints, set overall timeout to 60-90 seconds to account for retries. Streaming responses should have idle timeouts (no data received) rather than total timeouts - 10 seconds of no tokens indicates a stalled stream.

How do I test async LLM APIs?

Use pytest-asyncio with mock LLM responses. Test with pytest.mark.asyncio, mock the LLM client, and verify concurrent behavior with multiple simultaneous requests. Load test with tools like Locust configured for async endpoints. Our AI systems testing guide covers comprehensive testing strategies.

Can I cache LLM responses safely?

Yes, with caveats - cache only deterministic requests (temperature=0) or short-term cache (1-hour TTL) for repeated questions. Never cache streaming responses (cache only complete responses). Hash request parameters (model, messages, temperature) as cache key. Consider semantic caching where similar prompts return cached results.

What's the difference between FastAPI StreamingResponse and EventSourceResponse?

StreamingResponse is a generic streaming mechanism. EventSourceResponse (from sse-starlette) formats responses as Server-Sent Events with proper data: prefix and automatic keep-alive. For LLM APIs, use StreamingResponse with manual SSE formatting (as shown in this guide) for maximum control, or EventSourceResponse for convenience.

Conclusion

Building production async LLM APIs requires more than adding async keywords. You need concurrency control that actually covers the stream, a transport the client can consume, and failure handling that distinguishes transient from permanent errors.

  • Async is the throughput lever. One worker can hold hundreds of provider calls in flight as long as nothing on the path blocks the loop.
  • SSE is the default transport. Use fetch + ReadableStream on the client for POST bodies, and disable proxy buffering.
  • Hold the semaphore inside the generator. Otherwise your limiter is measuring the wrong thing.
  • Retry with jitter, break the circuit on outages, fall back to a second provider. Each layer covers a different failure duration.
  • Measure first-token latency and active streams, not just request counts; those are the metrics that move when something is wrong.

For the UX side of streaming (partial rendering, cancellation, resuming), continue with our guide to streaming LLM responses in production; for moving heavy work off the request path, see event-driven AI pipelines with Kafka and SQS.

Need help building or hardening an async LLM API? Talk to our backend engineering team or read about our backend API engineering services.

Free consultation

Book a free consultation call on async LLM APIs & FastAPI

30-minute call with the HinterBuild team. Discuss your project, architecture questions, or next steps — no obligation.

Book a meeting

Resources:

Keep reading

Related articles

AI Systems · 9 min read

Batch API for LLM Workloads: 50% Cost Savings on

Learn batch api for llm workloads through concrete architecture trade-offs, failure modes, rollout controls, and production measurement practices.

Read post

AI Systems · 12 min read

Streaming LLM Responses in Production

Streaming LLM Responses in Production guidance for engineers: compare architecture choices, avoid failure modes, and ship a measurable, reliable.

Read post

Backend Systems · 15 min read

API Versioning Strategies That Work in Production

API versioning strategies compared — URL, header, media type, and query versioning — with FastAPI and Go code, deprecation headers, and migration plans.

Read post

Backend Systems · 12 min read

gRPC vs REST vs GraphQL: How to Choose the Right API

Learn grpc vs rest vs graphql through concrete architecture trade-offs, failure modes, rollout controls, and production measurement practices.

Read post