Claude API Integration with FastAPI Backend

FastAPI has become a default choice for Python teams building AI-powered backends, and Claude is a natural fit for it: both are async-first, both handle JSON cleanly, and FastAPI's dependency injection makes it easy to keep API keys and client instances out of your route handlers. This guide walks through a working integration pattern — client setup, async request handling, streaming responses, and error handling — that you can drop into an existing FastAPI project.

The short answer to "how do I integrate Claude with FastAPI" is: use an async HTTP client (either the official Anthropic SDK or httpx), wrap it in a dependency, and expose it through your own endpoint so your frontend never talks to Claude's API directly. That last part matters for security (no API key in client-side code) and for consistency (you control retries, logging, and response shaping in one place).

Project structure

A minimal setup looks like this:

app/
  main.py
  claude_client.py
  routers/
    chat.py
  schemas.py

Keep the Claude client isolated in its own module so you can swap providers or add caching later without touching route logic.

Setting up the client

Install the SDK and configure it with your API key via environment variables — never hardcode it.

pip install anthropic fastapi uvicorn python-dotenv

# claude_client.py
import os
from anthropic import AsyncAnthropic

def get_claude_client() -> AsyncAnthropic:
    return AsyncAnthropic(api_key=os.environ["ANTHROPIC_API_KEY"])

Use AsyncAnthropic, not the sync client. FastAPI is built on an async event loop, and calling a blocking HTTP client inside an async def route will block the whole worker until the request completes, killing your concurrency.

A basic chat endpoint

# schemas.py
from pydantic import BaseModel

class ChatRequest(BaseModel):
    message: str
    model: str = "claude-sonnet-4-5"

class ChatResponse(BaseModel):
    reply: str
    input_tokens: int
    output_tokens: int

# routers/chat.py
from fastapi import APIRouter, Depends, HTTPException
from anthropic import AsyncAnthropic, APIError
from claude_client import get_claude_client
from schemas import ChatRequest, ChatResponse

router = APIRouter()

@router.post("/chat", response_model=ChatResponse)
async def chat(
    body: ChatRequest,
    client: AsyncAnthropic = Depends(get_claude_client),
):
    try:
        response = await client.messages.create(
            model=body.model,
            max_tokens=1024,
            messages=[{"role": "user", "content": body.message}],
        )
    except APIError as e:
        raise HTTPException(status_code=502, detail=str(e))

    return ChatResponse(
        reply=response.content[0].text,
        input_tokens=response.usage.input_tokens,
        output_tokens=response.usage.output_tokens,
    )

Using Depends for the client means you can override it in tests with a mock, which is far easier than patching a global instance.

Streaming responses through FastAPI

Chat UIs usually need token-by-token output. FastAPI supports this with StreamingResponse, and Anthropic's SDK exposes a streaming context manager that pairs with it cleanly:

from fastapi.responses import StreamingResponse

@router.post("/chat/stream")
async def chat_stream(
    body: ChatRequest,
    client: AsyncAnthropic = Depends(get_claude_client),
):
    async def event_generator():
        async with client.messages.stream(
            model=body.model,
            max_tokens=1024,
            messages=[{"role": "user", "content": body.message}],
        ) as stream:
            async for text in stream.text_stream:
                yield text

    return StreamingResponse(event_generator(), media_type="text/plain")

For a frontend expecting Server-Sent Events instead of a plain text stream, wrap each chunk in data: ...\n\n and set media_type="text/event-stream".

Handling tool use and structured responses

If your backend needs Claude to call functions — looking up a database record, hitting an internal service — define the tools in the request and check response.stop_reason for tool_use:

response = await client.messages.create(
    model=body.model,
    max_tokens=1024,
    tools=[{
        "name": "get_order_status",
        "description": "Look up an order by ID",
        "input_schema": {
            "type": "object",
            "properties": {"order_id": {"type": "string"}},
            "required": ["order_id"],
        },
    }],
    messages=[{"role": "user", "content": body.message}],
)

if response.stop_reason == "tool_use":
    tool_call = next(b for b in response.content if b.type == "tool_use")
    # run tool_call.name with tool_call.input, then send the result back

This pattern lets FastAPI act as the orchestrator: Claude decides what to call, your backend executes it against real data, and you send the result back in a follow-up message.

Error handling and rate limits

Claude's API returns standard HTTP status codes — 429 for rate limits, 529 for overload, 401 for auth issues. Wrap calls with retry logic for transient errors:

import asyncio
from anthropic import APIStatusError

async def call_with_retry(client, **kwargs):
    for attempt in range(3):
        try:
            return await client.messages.create(**kwargs)
        except APIStatusError as e:
            if e.status_code in (429, 529) and attempt < 2:
                await asyncio.sleep(2 ** attempt)
                continue
            raise

Log token usage from response.usage on every call. It's the cheapest way to catch a runaway prompt or a client sending oversized payloads before it shows up on your bill.

Simplifying the integration layer

If you'd rather not manage direct API keys, retries, and usage tracking yourself, SubToAPI sits between your FastAPI app and Claude: you get an application key (sub_live_...), the same request and streaming format, and a dashboard with usage metadata per key — useful when multiple services or team members share one Claude subscription. Swapping it in means changing the base URL and key in your claude_client.py, nothing else in your route logic changes. Check the quickstart and streaming docs if you want to compare request shapes before deciding.

Production checklist

• Use AsyncAnthropic, never the sync client, inside async routes
• Set explicit timeouts on the client to avoid hanging requests
• Validate and cap max_tokens server-side, don't trust client input
• Log usage.input_tokens and usage.output_tokens for cost visibility
• Return 502/503 to your own clients on upstream Claude errors, not raw tracebacks

FAQ

Should I call Claude's API directly from FastAPI or use a gateway? Calling it directly is fine for a single service with one API key. A gateway or provider like SubToAPI helps once you have multiple apps or team members sharing access and need separate keys, usage tracking, and centralized billing without each service holding the raw provider key.

Why does my FastAPI endpoint hang under load when calling Claude? This almost always means you're using the sync Anthropic client inside an async def route. Switch to AsyncAnthropic and await client.messages.create(...) so the event loop can serve other requests while waiting on the network call.

How do I stream Claude responses to a browser through FastAPI? Use the SDK's client.messages.stream() async context manager inside a generator function, then pass that generator to FastAPI's StreamingResponse. For SSE-compatible frontends, format each chunk as data: <text>\n\n and set the media type to text/event-stream.