Claude + FastAPI: 7 Agent-Patterns aus Produktion

Dieser Post ist eine technische Sammlung von Patterns fuer KI-Agenten auf Basis von Anthropic Claude und FastAPI, wie wir sie in unseren Kunden-Setups einsetzen. Kein Tutorial — eher eine Kurzreferenz der Entscheidungen, die in der Praxis den Unterschied zwischen „funktioniert in der Demo" und „funktioniert in Produktion" machen.

1. Claude-API aufrufen: streaming vs. batch

Die Default-Anleitungen zeigen fast immer messages.create() mit kompletter Response. Fuer User-facing Anwendungen ist das schlecht — das UI friert 3-15 Sekunden ein, bevor die Antwort kommt. Nutze stattdessen Server-Sent-Events.

from anthropic import AsyncAnthropic
from fastapi import APIRouter
from fastapi.responses import StreamingResponse

client = AsyncAnthropic()
router = APIRouter()

@router.post("/agent/chat")
async def chat(prompt: str):
    async def event_stream():
        async with client.messages.stream(
            model="claude-opus-4-6",
            max_tokens=4096,
            messages=[{"role": "user", "content": prompt}],
        ) as stream:
            async for chunk in stream.text_stream:
                yield f"data: {chunk}\n\n"
            yield "data: [DONE]\n\n"

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

Im Frontend (React/Next.js) nimmst du dann EventSource — sichtbare Tokens erscheinen sofort. Fuer Background-Jobs ohne UI (z.B. Batch-Processing) messages.create() mit Prompt-Caching.

2. Prompt-Caching nutzen (Cost- und Latenz-Spar)

Wenn ein grosser System-Prompt (System-Context, RAG-Chunks, Tool-Definitionen) ueber mehrere Requests identisch bleibt, nutze cache_control. Claude haelt das bis zu 5 Minuten, du zahlst nur 10 % vom normalen Input-Preis fuer Cache-Hits.

response = await client.messages.create(
    model="claude-opus-4-6",
    max_tokens=2048,
    system=[
        {
            "type": "text",
            "text": LANGES_SYSTEM_PROMPT,  # >1024 tokens
            "cache_control": {"type": "ephemeral"},
        }
    ],
    messages=[{"role": "user", "content": user_prompt}],
)

Im realen Einsatz bei uns: 73 % Kostensenkung bei Customer-Support-Bot, der die gleiche Dokumentation in jedem Request als Context mitgibt.

3. Tool-Use mit Pydantic-Validierung

Claude's Tool-Use ist maechtig, aber er kann Invalid-Arguments zurueckgeben („ich habe eine Idee, aber die JSON-Syntax stimmt nicht"). Defense: Pydantic-Models als Tool-Schema-Generator UND als Validator.

from pydantic import BaseModel, Field

class SearchTool(BaseModel):
    query: str = Field(description="Die Suchanfrage")
    limit: int = Field(default=10, ge=1, le=50)

tools = [
    {
        "name": "search_companies",
        "description": "Suche in der TechPuls-DB",
        "input_schema": SearchTool.model_json_schema(),
    }
]

# Bei jedem Tool-Call:
try:
    args = SearchTool(**tool_use.input)
    result = await do_search(args.query, args.limit)
except ValidationError as e:
    # Claude bekommt den Error zurueck und korrigiert sich selbst
    tool_result = {"error": str(e)}

Wichtig: Claude den Fehler zurueckspielen statt abzubrechen. Der Retry klappt erstaunlich oft beim naechsten Turn.

4. Rate-Limiting respektieren

Anthropic hat Per-Minute- und Per-Day-Limits. Im Produktionseinsatz fallen die schnell auf, wenn du Agent-Loops baust. Wir nutzen aiolimiter:

from aiolimiter import AsyncLimiter

# 50 Requests pro Minute, grosser Burst erlaubt
claude_limiter = AsyncLimiter(max_rate=50, time_period=60)

async def call_claude(messages):
    async with claude_limiter:
        return await client.messages.create(
            model="claude-opus-4-6",
            messages=messages,
            max_tokens=2048,
        )

Plus Retry-Logik mit tenacity fuer 529-Overload-Responses. Exponential backoff, max 3 Retries.

5. Structured Output ohne Tool-Use

Manchmal brauchst du strikt strukturierte JSON-Ausgabe, aber kein Tool-Use. Claude-Trick: prefill mit { starten — das zwingt Claude in JSON-Modus.

response = await client.messages.create(
    model="claude-opus-4-6",
    max_tokens=2048,
    messages=[
        {"role": "user", "content": "Extrahiere Name und Stadt aus: ..."},
        {"role": "assistant", "content": "{"},  # prefill
    ],
    stop_sequences=["\n\n"],
)

# Claudes Response ist jetzt der Rest der JSON-Struktur
json_str = "{" + response.content[0].text

Funktioniert 99 % der Zeit. Fuer die restlichen 1 %: Fallback mit Pydantic-Validation und einem Retry-Prompt („Nur valides JSON bitte").

6. Agent-Loops: Max-Iterations setzen

Ein Agent, der selbst entscheidet, wann er fertig ist, kann in Loops haengen. Immer harte Obergrenze.

MAX_ITERATIONS = 10

async def run_agent(user_prompt: str):
    messages = [{"role": "user", "content": user_prompt}]

    for iteration in range(MAX_ITERATIONS):
        response = await call_claude(messages)
        messages.append({"role": "assistant", "content": response.content})

        if response.stop_reason == "end_turn":
            return response.content

        # Tool-Use: Tools ausfuehren, Results zurueck an Claude
        tool_results = await execute_tools(response.content)
        messages.append({"role": "user", "content": tool_results})

    raise AgentIterationLimit(f"Agent hat {MAX_ITERATIONS} Iterations ueberschritten")

Logge dabei iteration und stop_reason — hilft beim Debuggen, warum ein Agent ge-looped hat.

7. Beobachtbarkeit: structlog + Anthropic-Request-IDs

Jeder Claude-Response kommt mit einer request_id, die Anthropic im Dashboard indexiert. Logge die immer.

import structlog

log = structlog.get_logger()

response = await client.messages.create(...)

log.info(
    "claude_response",
    request_id=response.id,  # fuer Anthropic-Support-Tickets
    model=response.model,
    stop_reason=response.stop_reason,
    input_tokens=response.usage.input_tokens,
    output_tokens=response.usage.output_tokens,
    cache_creation_tokens=response.usage.cache_creation_input_tokens,
    cache_read_tokens=response.usage.cache_read_input_tokens,
)

Bei Bugs haben wir oft rueck-verfolgen koennen, welcher Prompt zu welcher Response gefuehrt hat. Anthropic-Support kann Request-IDs auch nach mehreren Wochen noch lookupen.

Was wir nicht empfehlen

Langchain/LlamaIndex fuer Anthropic-Tools — zu viele Abstraktions-Schichten, das Debugging wird zur Hoelle. Nimm den Anthropic-SDK direkt.
Eigene Message-Serialisierung — das SDK macht das schon richtig, inkl. Pydantic-Models.
Response-Objekt-Weitergabe ueber Prozess-Grenzen — nicht pickle-bar. Stattdessen extrahiere die Felder, die du brauchst.

Beispiel-Setup auf GitHub

Wir haben ein Minimal-Template fuer Claude + FastAPI auf GitHub: github.com/TechLogia-de. Template mit streaming, caching, tool-use und structured logging.

Fragen zu Agent-Setups oder Code-Reviews: kontakt@techlogia.de.