Claude API FastAPI Integration Tutorial

Claude API FastAPI Integration Tutorial

If you're building a Python backend and want to wire Claude into it, FastAPI is a natural fit: async-first, fast to iterate on, and good at handling streaming responses without blocking your event loop. This tutorial walks through a complete integration — from the first request to a production-ready endpoint with streaming, error handling, and tool use.

The short version: you create an async route, call Claude's messages endpoint with httpx (or an SDK), and return the response as JSON or as a server-sent event stream. Everything below expands on that with working code you can drop into a real project.

Project setup

Start with a minimal FastAPI app and the dependencies you'll need:

pip install fastapi uvicorn httpx python-dotenv

Project structure:

app/
  main.py
  claude_client.py
.env

Store your API key in .env and load it with python-dotenv. Never hardcode keys in source files, and never commit .env to git.

# .env
CLAUDE_API_KEY=your_key_here

Step 1: A thin Claude client

Keep the HTTP call logic separate from your route handlers. This makes it easier to swap providers, add retries, or mock requests in tests.

# app/claude_client.py
import os
import httpx

API_URL = "https://api.anthropic.com/v1/messages"
API_KEY = os.getenv("CLAUDE_API_KEY")

async def call_claude(messages: list, model: str = "claude-3-5-sonnet-20241022", max_tokens: int = 1024):
    headers = {
        "x-api-key": API_KEY,
        "anthropic-version": "2023-06-01",
        "content-type": "application/json",
    }
    payload = {
        "model": model,
        "max_tokens": max_tokens,
        "messages": messages,
    }
    async with httpx.AsyncClient(timeout=30) as client:
        response = await client.post(API_URL, headers=headers, json=payload)
        response.raise_for_status()
        return response.json()

Using httpx.AsyncClient instead of requests matters here — a synchronous call inside an async route will block the whole event loop and tank throughput under load.

Step 2: The FastAPI route

# app/main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from app.claude_client import call_claude

app = FastAPI()

class ChatRequest(BaseModel):
    prompt: str

@app.post("/chat")
async def chat(req: ChatRequest):
    try:
        result = await call_claude([
            {"role": "user", "content": req.prompt}
        ])
        text = result["content"][0]["text"]
        return {"reply": text}
    except Exception as e:
        raise HTTPException(status_code=502, detail=str(e))

Run it with:

uvicorn app.main:app --reload

That's a working integration. Everything else in this guide is about making it production-grade.

Step 3: Streaming responses

For chat UIs, users expect tokens to appear as they're generated, not all at once after a multi-second wait. FastAPI's StreamingResponse handles this cleanly when paired with Claude's streaming mode.

from fastapi.responses import StreamingResponse
import httpx, json, os

async def stream_claude(prompt: str):
    headers = {
        "x-api-key": os.getenv("CLAUDE_API_KEY"),
        "anthropic-version": "2023-06-01",
        "content-type": "application/json",
    }
    payload = {
        "model": "claude-3-5-sonnet-20241022",
        "max_tokens": 1024,
        "stream": True,
        "messages": [{"role": "user", "content": prompt}],
    }
    async with httpx.AsyncClient(timeout=None) as client:
        async with client.stream("POST", "https://api.anthropic.com/v1/messages",
                                  headers=headers, json=payload) as response:
            async for line in response.aiter_lines():
                if line.startswith("data:"):
                    yield line[5:].strip() + "\n"

@app.post("/chat/stream")
async def chat_stream(req: ChatRequest):
    return StreamingResponse(stream_claude(req.prompt), media_type="text/event-stream")

On the client side, you consume this as a normal SSE stream — EventSource in the browser or any SSE-aware fetch wrapper.

Step 4: Tool use (function calling)

Claude's tool use lets the model call functions you define — database lookups, calculators, internal APIs — and return structured results. In FastAPI, define the tool schema alongside your route and dispatch based on the model's response:

tools = [{
    "name": "get_order_status",
    "description": "Look up an order's status by ID",
    "input_schema": {
        "type": "object",
        "properties": {"order_id": {"type": "string"}},
        "required": ["order_id"],
    },
}]

async def call_claude_with_tools(messages: list):
    payload = {
        "model": "claude-3-5-sonnet-20241022",
        "max_tokens": 1024,
        "messages": messages,
        "tools": tools,
    }
    # same httpx call as before, inspect response["content"] for tool_use blocks

When the response contains a tool_use block, your FastAPI handler executes the corresponding function, appends the result as a tool_result message, and sends the conversation back to Claude for a final answer. This round-trip pattern is the core of any agentic FastAPI integration.

Step 5: Error handling and timeouts

Claude's API can return rate limit errors (429), overload errors (529), and validation errors (400). Wrap calls with explicit handling rather than letting httpx exceptions bubble up as generic 500s:

try:
    response = await client.post(API_URL, headers=headers, json=payload)
    response.raise_for_status()
except httpx.HTTPStatusError as e:
    if e.response.status_code == 429:
        raise HTTPException(status_code=429, detail="Rate limited, retry shortly")
    raise HTTPException(status_code=502, detail="Claude API error")
except httpx.TimeoutException:
    raise HTTPException(status_code=504, detail="Claude API timeout")

Add exponential backoff for 429/529 responses if your app sees meaningful traffic — a simple retry loop with jitter is usually enough.

Where SubToAPI fits

If you'd rather not manage raw Anthropic credentials, rate-limit logic, and per-key usage tracking yourself, SubToAPI wraps the same request/response shape behind an HTTPS API with its own application keys (sub_live_...), streaming, tool use, and usage metadata in a dashboard. Your FastAPI client code barely changes — you just point httpx at https://api.subtoapi.app/v1/messages with Authorization: Bearer $SUBTOAPI_KEY instead of managing provider keys directly:

curl https://api.subtoapi.app/v1/messages \
  -H "Authorization: Bearer $SUBTOAPI_KEY" \
  -H "content-type: application/json" \
  -d '{"model":"claude-3-5-sonnet-20241022","max_tokens":1024,"messages":[{"role":"user","content":"Hello"}]}'

This is useful when multiple services or team members need access without sharing a single raw API key. See the quickstart and messages docs for the full request format, or the streaming docs if you're implementing Step 3 above against SubToAPI instead of calling Anthropic directly.

Questions

Do I need the official Anthropic SDK for FastAPI integration? No. httpx with async support is enough and gives you more control over timeouts, retries, and streaming. The SDK is convenient but adds a dependency you don't strictly need.

How do I keep Claude calls from blocking other requests? Always use an async HTTP client (httpx.AsyncClient), never a synchronous one, inside async def routes. A blocking call will stall FastAPI's single event loop for every other request.

Can I run this integration behind a gateway instead of calling Anthropic directly? Yes — the request/response shape stays the same whether you call Anthropic's endpoint or a service like SubToAPI that sits in front of it with its own key management and usage tracking.