Claude API FastAPI Integration Example

If you're building a Python backend with FastAPI and want to call Claude, the fastest path is a small async wrapper around the HTTP API using httpx, exposed through your own FastAPI route. This article walks through a full working example: project setup, a streaming endpoint, error handling, and how to structure it so it survives contact with production traffic.

The core pattern is always the same: FastAPI handles the incoming request and validation, an async HTTP client calls Claude (either Anthropic's API directly or a gateway like SubToAPI), and the response is either returned as JSON or streamed back to the client via Server-Sent Events. Below is a minimal but realistic implementation you can copy and adapt.

Project setup

pip install fastapi uvicorn httpx python-dotenv

Create a .env file:

CLAUDE_API_KEY=sub_live_your_key_here
CLAUDE_API_BASE=https://api.subtoapi.app/v1

Using a key issued per application (rather than sharing a single raw Anthropic key across services) makes it much easier to rotate credentials and track usage per integration — this is one of the things SubToAPI's dashboard handles for you out of the box, alongside the HTTPS API itself.

A basic FastAPI route

import os
import httpx
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

CLAUDE_API_BASE = os.environ["CLAUDE_API_BASE"]
CLAUDE_API_KEY = os.environ["CLAUDE_API_KEY"]

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

@app.post("/chat")
async def chat(req: ChatRequest):
    async with httpx.AsyncClient(timeout=60.0) as client:
        try:
            resp = await client.post(
                f"{CLAUDE_API_BASE}/messages",
                headers={
                    "Authorization": f"Bearer {CLAUDE_API_KEY}",
                    "Content-Type": "application/json",
                },
                json={
                    "model": req.model,
                    "max_tokens": req.max_tokens,
                    "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.RequestError:
            raise HTTPException(status_code=502, detail="Upstream request failed")

    data = resp.json()
    return {"reply": data["content"][0]["text"]}

This gives you a working /chat endpoint. Run it with:

uvicorn main:app --reload

And test it:

curl -X POST http://localhost:8000/chat \
  -H "Content-Type: application/json" \
  -d '{"message": "Explain async/await in Python in two sentences"}'

Full request/response schema details, including how multi-turn conversations and system prompts work, are covered in /docs/messages.

Streaming responses through FastAPI

For chat UIs, you usually want to stream tokens as they arrive instead of waiting for the full response. FastAPI supports this via StreamingResponse, and httpx supports streaming reads. Here's the pattern:

from fastapi.responses import StreamingResponse
import json

@app.post("/chat/stream")
async def chat_stream(req: ChatRequest):
    async def event_generator():
        async with httpx.AsyncClient(timeout=None) as client:
            async with client.stream(
                "POST",
                f"{CLAUDE_API_BASE}/messages",
                headers={
                    "Authorization": f"Bearer {CLAUDE_API_KEY}",
                    "Content-Type": "application/json",
                },
                json={
                    "model": req.model,
                    "max_tokens": req.max_tokens,
                    "stream": True,
                    "messages": [{"role": "user", "content": req.message}],
                },
            ) as response:
                async for line in response.aiter_lines():
                    if line.startswith("data:"):
                        payload = line[len("data:"):].strip()
                        if payload and payload != "[DONE]":
                            yield f"data: {payload}\n\n"

    return StreamingResponse(event_generator(), media_type="text/event-stream")

On the frontend, consume this with EventSource or a fetch call that reads the response body as a stream. The mechanics of Claude's streaming event format (message start, content block deltas, message stop) are documented in /docs/streaming — worth reading before you build retry logic around partial chunks.

Handling tool use from FastAPI

If your FastAPI service needs Claude to call functions (database lookups, internal APIs, calculations), pass a tools array in the request body and check the response for tool_use content blocks:

@app.post("/chat/tools")
async def chat_with_tools(req: ChatRequest):
    tools = [
        {
            "name": "get_order_status",
            "description": "Look up the status of an order by ID",
            "input_schema": {
                "type": "object",
                "properties": {"order_id": {"type": "string"}},
                "required": ["order_id"],
            },
        }
    ]

    async with httpx.AsyncClient(timeout=60.0) as client:
        resp = await client.post(
            f"{CLAUDE_API_BASE}/messages",
            headers={"Authorization": f"Bearer {CLAUDE_API_KEY}"},
            json={
                "model": req.model,
                "max_tokens": req.max_tokens,
                "tools": tools,
                "messages": [{"role": "user", "content": req.message}],
            },
        )
        data = resp.json()

    for block in data.get("content", []):
        if block.get("type") == "tool_use":
            # run your actual lookup here, then send a follow-up
            # message with a tool_result block
            return {"tool_call": block["name"], "input": block["input"]}

    return {"reply": data["content"][0]["text"]}

You'd then run the real function, append a tool_result message, and call /messages again to get Claude's final answer. The full request shape for tool schemas and multi-step tool loops is in /docs/tools.

Why route through a gateway instead of calling Anthropic directly

You can point CLAUDE_API_BASE straight at Anthropic's endpoint and it'll work identically — the code above doesn't change. The reason teams put a layer like SubToAPI in between is operational: issuing a separate sub_live_... key per service or environment, seeing usage metadata per key in one dashboard, and managing team seats without sharing a single credential across every FastAPI deployment. If you're just getting started, /docs/quickstart walks through getting a key and making your first call in under five minutes, and /pricing covers the Solo, Team, and Scale plans along with the free trial.

Production checklist

• Set timeout explicitly on your httpx.AsyncClient — Claude responses for long generations can take tens of seconds.
• Catch httpx.HTTPStatusError separately from httpx.RequestError so you can distinguish upstream 4xx/5xx from network failures.
• Keep the API key out of source control; load it from environment variables, not hardcoded strings.
• For streaming endpoints, disable response buffering in your reverse proxy (e.g., proxy_buffering off in nginx) or tokens will arrive in bursts instead of smoothly.

questions

Do I need the official Anthropic SDK to use FastAPI with Claude? No. A plain httpx.AsyncClient making POST requests to the Messages endpoint is enough and keeps your dependency footprint small — the SDK is a convenience wrapper, not a requirement.

Can I stream Claude's response through FastAPI to a browser? Yes, using StreamingResponse with media_type="text/event-stream", reading the upstream response with client.stream() and forwarding each data: line as it arrives.

Why would I use SubToAPI instead of calling Claude directly from FastAPI? The integration code is identical either way — you just change the base URL and key. SubToAPI adds per-application keys, usage dashboards, and team seat management on top, which direct API access doesn't provide.