Claude API Integration with FastAPI: A Python Guide

Integrating the Claude API with FastAPI means wiring up an async HTTP client to Anthropic's Messages API, exposing it through your own endpoints, and handling streaming, errors, and authentication correctly so your service doesn't fall over under real traffic. This guide walks through a working setup from scratch: project structure, request/response models, streaming with Server-Sent Events, and the production details that tutorials usually skip.

FastAPI is a natural fit for Claude because both are async-first. You get non-blocking request handling on your server while Claude generates tokens, which matters a lot once you have concurrent users hitting a chat endpoint.

Setting Up the Project

Start with a minimal dependency set:

pip install fastapi uvicorn httpx python-dotenv

You don't need Anthropic's official SDK to make this work — a plain httpx.AsyncClient against the REST API is enough, and it keeps your dependency surface small if you're proxying requests through a service like SubToAPI or switching providers later.

# main.py
import os
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import httpx

app = FastAPI()

CLAUDE_API_KEY = os.environ["ANTHROPIC_API_KEY"]
CLAUDE_URL = "https://api.anthropic.com/v1/messages"

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

class ChatResponse(BaseModel):
    reply: str

A Basic Endpoint

The core pattern: build the request payload, call the API asynchronously, parse the response, return clean JSON to your frontend.

@app.post("/chat", response_model=ChatResponse)
async def chat(req: ChatRequest):
    async with httpx.AsyncClient(timeout=30.0) as client:
        try:
            resp = await client.post(
                CLAUDE_URL,
                headers={
                    "x-api-key": CLAUDE_API_KEY,
                    "anthropic-version": "2023-06-01",
                    "content-type": "application/json",
                },
                json={
                    "model": req.model,
                    "max_tokens": 1024,
                    "messages": [{"role": "user", "content": req.message}],
                },
            )
            resp.raise_for_status()
        except httpx.HTTPStatusError as e:
            raise HTTPException(status_code=e.response.status_code, detail=e.response.text)
        except httpx.TimeoutException:
            raise HTTPException(status_code=504, detail="Claude API timeout")

    data = resp.json()
    text = "".join(block["text"] for block in data["content"] if block["type"] == "text")
    return ChatResponse(reply=text)

A few things worth noting here: raise_for_status() catches 4xx/5xx responses so you're not silently returning empty replies to users. The timeout is explicit rather than relying on defaults, because Claude responses on longer prompts can take several seconds.

Streaming Responses to the Frontend

Blocking on a full generation before responding feels slow for chat-style UIs. FastAPI supports streaming via StreamingResponse, and Claude supports server-sent events natively when you set "stream": true.

from fastapi.responses import StreamingResponse
import json

async def stream_claude(message: str, model: str):
    async with httpx.AsyncClient(timeout=None) as client:
        async with client.stream(
            "POST",
            CLAUDE_URL,
            headers={
                "x-api-key": CLAUDE_API_KEY,
                "anthropic-version": "2023-06-01",
                "content-type": "application/json",
            },
            json={
                "model": model,
                "max_tokens": 1024,
                "stream": True,
                "messages": [{"role": "user", "content": message}],
            },
        ) as response:
            async for line in response.aiter_lines():
                if line.startswith("data:"):
                    payload = line[5:].strip()
                    if payload and payload != "[DONE]":
                        chunk = json.loads(payload)
                        if chunk.get("type") == "content_block_delta":
                            yield chunk["delta"].get("text", "")

@app.post("/chat/stream")
async def chat_stream(req: ChatRequest):
    return StreamingResponse(
        stream_claude(req.message, req.model),
        media_type="text/plain",
    )

This yields text chunks as they arrive, so your frontend can render them incrementally instead of waiting for the full response.

Handling Errors and Rate Limits

Claude's API returns structured error bodies with a type field (rate_limit_error, overloaded_error, invalid_request_error, etc.). It's worth mapping these explicitly rather than passing through a generic 500:

def parse_claude_error(body: dict) -> tuple[int, str]:
    error_type = body.get("error", {}).get("type", "unknown")
    mapping = {
        "rate_limit_error": (429, "Rate limit exceeded, retry with backoff"),
        "overloaded_error": (503, "Claude is temporarily overloaded"),
        "invalid_request_error": (400, "Invalid request to Claude API"),
    }
    return mapping.get(error_type, (500, "Unexpected Claude API error"))

Wire this into your except httpx.HTTPStatusError block so your API consumers get actionable status codes instead of opaque failures.

Managing API Keys and Environments

Don't hardcode keys or check them into source control. Load them via python-dotenv in development and environment variables in production:

from dotenv import load_dotenv
load_dotenv()

If you're building a product on top of Claude — rather than a single internal tool — you'll also want per-user API keys, usage tracking, and team access without building that billing and auth layer yourself. This is exactly the gap SubToAPI fills: it turns your existing Claude access into a proper HTTPS API with scoped sub_live_... keys, streaming, tool use, and usage metadata already built in. If your FastAPI app just needs to call Claude reliably without owning key management and quota logic, check the quickstart — it's a drop-in replacement for calling Anthropic directly, using the same request shape shown in the messages docs.

Adding Tool Use

FastAPI endpoints are also a good place to expose tool-calling to Claude, since your route handlers are already async functions that can hit databases or external APIs. Define your tool schema in the request payload, then dispatch on tool_use blocks in the response and feed results back in a follow-up call. See the tools docs for the exact schema if you're doing this against SubToAPI instead of the raw Anthropic endpoint — the request/response format matches Anthropic's, so switching is a config change, not a rewrite.

Deployment Notes

Run with uvicorn main:app --workers 4 behind a reverse proxy in production, and keep httpx.AsyncClient instances scoped per-request or use a shared client with connection pooling via FastAPI's lifespan events to avoid socket exhaustion under load. Set max_tokens conservatively for chat endpoints — it directly affects latency and cost, and unbounded values are a common cause of runaway response times.

questions

Do I need Anthropic's official Python SDK to use Claude with FastAPI? No. httpx.AsyncClient against the REST endpoint works fine and keeps your async code path consistent with the rest of FastAPI, though the SDK is a reasonable choice if you want built-in retries.

How do I stream Claude responses through FastAPI to a browser? Use StreamingResponse with an async generator that reads Claude's SSE stream and yields text deltas as they arrive, as shown above.

Can I use FastAPI with SubToAPI instead of calling Anthropic directly? Yes — SubToAPI exposes the same Messages-style API shape at api.subtoapi.app/v1/..., so your FastAPI client code changes only the base URL and auth header. See pricing and signup for details.