FastAPI dependency injection for LLM client management

Most LLM backends bolt a client instance onto a global variable and call it a day. That breaks the moment you need per-tenant credentials, fallback routing, or test isolation. Using fastapi dependency injection llm client patterns correctly gives you explicit lifecycles, clean overrides, and a single place to enforce rate limits and cache hints.

Start with a narrow client interface

Don’t inject the SDK class directly. Define a protocol that matches the calls your app actually makes. This keeps your endpoints agnostic to whether you’re calling OpenAI, a local vLLM instance, or a gateway.

from abc import ABC, abstractmethod
from dataclasses import dataclass

@dataclass
class CompletionResult:
    text: str
    model: str
    usage_tokens: int

class LLMClient(ABC):
    @abstractmethod
    async def complete(self, prompt: str, model: str, **opts) -> CompletionResult: ...

A thin abstract base forces you to decide what fields matter (token counts, model id) up front. Later you can swap implementations without touching route handlers.

Build the HTTP implementation once

Use httpx.AsyncClient as the transport. Never instantiate it per request—connection pooling is the entire point.

import httpx

class GatewayLLMClient(LLMClient):
    def __init__(self, base_url: str, api_key: str, timeout: float = 30.0):
        self._client = httpx.AsyncClient(
            base_url=base_url,
            headers={"Authorization": f"Bearer {api_key}"},
            timeout=timeout,
        )

    async def complete(self, prompt: str, model: str, **opts) -> CompletionResult:
        resp = await self._client.post(
            "/v1/completions",
            json={"model": model, "prompt": prompt, **opts}
        )
        resp.raise_for_status()
        data = resp.json()
        return CompletionResult(
            text=data["choices"][0]["text"],
            model=data["model"],
            usage_tokens=data["usage"]["total_tokens"],
        )

    async def aclose(self) -> None:
        await self._client.aclose()

Attach the client to app state via lifespan

FastAPI’s lifespan context is the correct place to create and tear down the singleton. The fastapi dependency injection llm client wiring then becomes a one-line getter.

from contextlib import asynccontextmanager
from fastapi import FastAPI, Request

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.llm = GatewayLLMClient(
        base_url="https://api.example.com",
        api_key="sk-env",
    )
    yield
    await app.state.llm.aclose()

app = FastAPI(lifespan=lifespan)

def get_llm_client(request: Request) -> LLMClient:
    return request.app.state.llm

Endpoints declare the dependency and stay readable:

@app.post("/summarize")
async def summarize(text: str, llm: LLMClient = Depends(get_llm_client)):
    result = await llm.complete(prompt=text, model="gpt-4o-mini")
    return {"summary": result.text}

Handle multi-tenant credentials with request-scoped factories

Sharing one client across all tenants is fine when the gateway handles auth. If you must isolate by API key, build a pool keyed by tenant id. Don’t recreate the underlying HTTP client each call—reuse a per-tenant client stored in a dict.

from fastapi import Header

_tenant_clients: dict[str, GatewayLLMClient] = {}

def get_tenant_llm(request: Request, tenant_key: str = Header("X-Tenant-Key")) -> LLMClient:
    if tenant_key not in _tenant_clients:
        _tenant_clients[tenant_key] = GatewayLLMClient(
            base_url="https://api.example.com",
            api_key=tenant_key,
        )
    return _tenant_clients[tenant_key]

Tradeoff: memory grows with tenant count. Add TTL eviction if you have thousands of keys.

Forward routing and cache hints cleanly

When you proxy to an inference gateway, routing directives belong in the dependency, not the endpoint. If you route through a gateway such as n4n.ai, the OpenAI-compatible endpoint addresses 240+ models and honors client routing directives; your dependency can forward provider cache-control hints by passing them through to the request without extra branching.

Extend the interface to accept a routing dict:

class LLMClient(ABC):
    @abstractmethod
    async def complete(self, prompt: str, model: str, routing: dict | None = None, **opts) -> CompletionResult: ...

Implementation maps routing to headers:

    async def complete(self, prompt: str, model: str, routing=None, **opts):
        headers = {}
        if routing and "cache" in routing:
            headers["Cache-Control"] = routing["cache"]
        resp = await self._client.post(
            "/v1/completions",
            json={"model": model, "prompt": prompt, **opts},
            headers=headers,
        )

Now the endpoint stays dumb: await llm.complete(prompt, model, routing={"cache": "max-age=3600"}).

Override dependencies for tests

The biggest win of fastapi dependency injection llm client design is app.dependency_overrides. You get hermetic tests without mocking httpx.

class FakeLLMClient(LLMClient):
    async def complete(self, prompt: str, model: str, routing=None, **opts) -> CompletionResult:
        return CompletionResult(text="fake", model=model, usage_tokens=10)

app.dependency_overrides[get_llm_client] = lambda: FakeLLMClient()

def test_summarize():
    client = TestClient(app)
    resp = client.post("/summarize", json={"text": "hello"})
    assert resp.status_code == 200
    assert resp.json()["summary"] == "fake"

Reset overrides after the suite to avoid leakage.

Pitfalls that will bite you

Closing the client too early

If you create the AsyncClient inside a dependency without a lifespan, pytest may close it before the request finishes. Always bind long-lived transports to app.state.

Mixing sync SDKs in async routes

OpenAI’s old sync client blocks the event loop. Use the async version or run it in a thread pool. Blocking a route under load cascades into 502s.

Ignoring token metering

If your gateway returns per-token usage, surface it. Add a middleware that logs CompletionResult.usage_tokens per tenant. Without it you can’t debug cost spikes.

Hidden global state

The _tenant_clients dict above is a module global. In serverless environments with multiple worker reloads, it won’t be shared. Prefer a Redis-backed pool or accept that each worker caches independently.

Tradeoffs: thin wrapper vs full SDK passthrough

A purist wraps every model capability behind your interface. That’s clean for tests but lags when the provider adds a new parameter—you edit the ABC, the impl, and every caller.

A pragmatic middle ground: keep the ABC minimal (complete, chat) but let **opts pass through unknown fields. You lose some type safety but ship faster. For most internal backends, that’s correct.

If you need streaming, add a separate stream method returning an async iterator. Don’t overload complete with a stream=True flag; it complicates the return type and breaks the fake client.

Put it together: ordered path

Define LLMClient ABC with the two or three methods you actually call.
Implement against httpx.AsyncClient; never instantiate per request.
Create the singleton in lifespan and expose via get_llm_client(request).
For multi-tenant, return pooled clients keyed by header; evict on TTL.
Pass routing/cache hints as a parameter mapped to headers in the impl.
Write endpoint handlers that depend on the abstraction, not the SDK.
Use dependency_overrides with a fake in tests; assert on usage tokens.
Add middleware to record token counts per route and tenant.

Following this, your fastapi dependency injection llm client setup stays debuggable when a provider degrades, and you can swap the backend by writing one new class.

When not to use DI

If your service makes a single LLM call at startup and exits, a module-level client is simpler. Dependency injection pays off when you have live routes, multiple environments, and a test suite that must run without network access. Don’t add the pattern purely because the blogosphere says so—add it when the lifecycle complexity is real.