Provider SDK integration patterns: OpenAI and Anthropic
Why this matters
Most tutorials show you how to make one API call. Production systems need more: a client that handles authentication securely, retries the right errors, surfaces actionable logs, normalizes response shapes across providers, and does not require refactoring your application layer when a provider releases a breaking SDK change.
This lesson covers the patterns that distinguish a maintainable provider integration from an ad-hoc one.
Core concepts
Choosing sync vs. async clients. Both OpenAI and Anthropic provide sync and async Python clients. The sync client blocks the thread; the async client integrates with asyncio. Use async for:
- FastAPI handlers and any async web framework
- Batch processing where you want concurrent requests
- Any context where event loops are already running
Use sync for:
- CLI scripts and one-off pipelines
- Worker threads in a process pool
- Celery tasks that run in a non-async context
Mixing them carelessly causes RuntimeError: This event loop is already running. If you are in an async context, use the async client throughout.
Authentication and secret management. Provider keys should never live in source code. The standard pattern:
import os
from openai import AsyncOpenAI
from anthropic import AsyncAnthropic
# Both SDKs read from env automatically if the key is not passed explicitly
openai_client = AsyncOpenAI() # reads OPENAI_API_KEY
anthropic_client = AsyncAnthropic() # reads ANTHROPIC_API_KEY
# Or explicitly:
openai_client = AsyncOpenAI(api_key=os.environ["OPENAI_API_KEY"])
In production, use a secrets manager (AWS Secrets Manager, GCP Secret Manager, Doppler) to inject the key at runtime. Never commit .env files to source control.
OpenAI async call pattern with token tracking.
import time
from openai import AsyncOpenAI
from dataclasses import dataclass
client = AsyncOpenAI()
@dataclass
class LLMResult:
text: str
model: str
input_tokens: int
output_tokens: int
latency_ms: int
async def openai_generate(
messages: list[dict],
model: str = "gpt-4o-mini",
max_tokens: int = 1024,
temperature: float = 0.7,
) -> LLMResult:
start = time.monotonic()
response = await client.chat.completions.create(
model=model,
messages=messages,
max_tokens=max_tokens,
temperature=temperature,
)
return LLMResult(
text=response.choices[0].message.content or "",
model=response.model,
input_tokens=response.usage.prompt_tokens,
output_tokens=response.usage.completion_tokens,
latency_ms=int((time.monotonic() - start) * 1000),
)
Anthropic async call pattern. Anthropic's API separates the system prompt from the messages array:
from anthropic import AsyncAnthropic
client = AsyncAnthropic()
async def anthropic_generate(
system: str,
messages: list[dict],
model: str = "claude-haiku-4-5",
max_tokens: int = 1024,
) -> LLMResult:
start = time.monotonic()
response = await client.messages.create(
model=model,
system=system,
messages=messages,
max_tokens=max_tokens,
)
return LLMResult(
text=response.content[0].text if response.content else "",
model=response.model,
input_tokens=response.usage.input_tokens,
output_tokens=response.usage.output_tokens,
latency_ms=int((time.monotonic() - start) * 1000),
)
The key difference: OpenAI accepts the system prompt as {"role": "system", "content": "..."} inside the messages array. Anthropic accepts it as a top-level system parameter and expects the messages array to contain only user and assistant roles.
Streaming responses. For user-facing text generation, streaming is a UX requirement. Both SDKs support it:
# OpenAI streaming
async def openai_stream(messages: list[dict], model: str = "gpt-4o-mini"):
async with client.chat.completions.stream(
model=model,
messages=messages,
max_tokens=1024,
) as stream:
async for text in stream.text_stream:
yield text # yield each token chunk to caller
# Anthropic streaming
async def anthropic_stream(system: str, messages: list[dict], model: str = "claude-haiku-4-5"):
async with anthropic_client.messages.stream(
model=model,
system=system,
messages=messages,
max_tokens=1024,
) as stream:
async for text in stream.text_stream:
yield text
Provider-agnostic wrapper. Build an abstraction layer so your application code never imports provider SDKs directly:
from abc import ABC, abstractmethod
from typing import AsyncIterator
class ProviderClient(ABC):
@abstractmethod
async def generate(
self,
system: str,
messages: list[dict],
model: str,
max_tokens: int = 1024,
) -> LLMResult: ...
@abstractmethod
async def stream(
self,
system: str,
messages: list[dict],
model: str,
max_tokens: int = 1024,
) -> AsyncIterator[str]: ...
class OpenAIClient(ProviderClient):
def __init__(self, model: str = "gpt-4o-mini") -> None:
self._client = AsyncOpenAI()
self._default_model = model
async def generate(self, system: str, messages: list[dict], model: str = "", max_tokens: int = 1024) -> LLMResult:
full_messages = [{"role": "system", "content": system}, *messages]
return await openai_generate(full_messages, model=model or self._default_model, max_tokens=max_tokens)
async def stream(self, system: str, messages: list[dict], model: str = "", max_tokens: int = 1024) -> AsyncIterator[str]:
full_messages = [{"role": "system", "content": system}, *messages]
async for chunk in openai_stream(full_messages, model=model or self._default_model):
yield chunk
class AnthropicClient(ProviderClient):
def __init__(self, model: str = "claude-haiku-4-5") -> None:
self._client = AsyncAnthropic()
self._default_model = model
async def generate(self, system: str, messages: list[dict], model: str = "", max_tokens: int = 1024) -> LLMResult:
return await anthropic_generate(system, messages, model=model or self._default_model, max_tokens=max_tokens)
async def stream(self, system: str, messages: list[dict], model: str = "", max_tokens: int = 1024) -> AsyncIterator[str]:
async for chunk in anthropic_stream(system, messages, model=model or self._default_model):
yield chunk
Application code depends on ProviderClient. Swapping providers is one injection point change. Testing uses a MockProviderClient that never touches the network.
Error handling normalization. Provider SDK exceptions are provider-specific. Normalize them at the adapter boundary:
from openai import RateLimitError as OpenAIRateLimit, APIStatusError as OpenAIStatus
from anthropic import RateLimitError as AnthropicRateLimit, APIStatusError as AnthropicStatus
class ProviderError(Exception):
def __init__(self, message: str, retryable: bool, status_code: int | None = None):
super().__init__(message)
self.retryable = retryable
self.status_code = status_code
def normalize_openai_error(exc: Exception) -> ProviderError:
if isinstance(exc, OpenAIRateLimit):
return ProviderError("Rate limit exceeded", retryable=True, status_code=429)
if isinstance(exc, OpenAIStatus):
return ProviderError(str(exc), retryable=exc.status_code >= 500, status_code=exc.status_code)
return ProviderError(str(exc), retryable=False)
def normalize_anthropic_error(exc: Exception) -> ProviderError:
if isinstance(exc, AnthropicRateLimit):
return ProviderError("Rate limit exceeded", retryable=True, status_code=429)
if isinstance(exc, AnthropicStatus):
return ProviderError(str(exc), retryable=exc.status_code >= 500, status_code=exc.status_code)
return ProviderError(str(exc), retryable=False)
Downstream retry logic only needs to inspect ProviderError.retryable. It never imports provider-specific exception classes.
Common mistakes
Importing the SDK client at module level in tests. Module-level client = AsyncOpenAI() runs during import, which reads OPENAI_API_KEY from the environment. Tests that do not set this variable fail at import time, not at test execution time. Defer client creation to a factory function or use dependency injection.
Using the sync client in async handlers. A FastAPI route that calls client.chat.completions.create(...) (sync) blocks the event loop thread, degrading throughput for all concurrent requests. Always use the async client in async contexts.
Not closing the async client. AsyncOpenAI and AsyncAnthropic manage an internal httpx.AsyncClient. In long-running services, close it on shutdown to avoid resource leaks:
@asynccontextmanager
async def lifespan(app):
yield
await openai_client.close()
await anthropic_client.close()
Hardcoding model names in business logic. When gpt-4o-mini is referenced in 15 places and a new model is released, you have 15 changes to make. Store model names in config. Pass them as parameters.
Try it yourself
- Wire up both provider clients in a FastAPI app with a
/generateendpoint. Pass aproviderquery parameter that routes the request to OpenAI or Anthropic. Return theLLMResultas JSON. - Add a streaming endpoint
/streamthat uses server-sent events (SSE) to forward tokens to the browser as they arrive. UseStreamingResponsein FastAPI. - Write a
MockProviderClientthat returns a fixed response and records every call. Use it in a pytest fixture so your route tests never touch real provider APIs.