FastAPI + Anthropic: Building Reliable Multi-Step AI Workflows with Claude

A single Claude API call is straightforward. A workflow that chains several calls together — using tool calls, retrieved context, and conditional branching — has failure modes a simple chat endpoint doesn't: a malformed tool call, a step that times out, or token costs that balloon silently across a long chain. This post covers the patterns that keep multi-step Claude workflows reliable in production.

Structuring a multi-step call chain

Treat each step as its own function with its own error boundary, rather than one long prompt trying to do everything at once. Smaller, focused steps are easier to validate and retry independently.

async def run_research_workflow(query: str) -> WorkflowResult:
    total_tokens = 0

    plan = await anthropic_client.messages.create(
        model="claude-3-5-sonnet-20241022",
        messages=[{"role": "user", "content": build_plan_prompt(query)}],
    )
    total_tokens += plan.usage.input_tokens + plan.usage.output_tokens

    context = await rag.query(question=query, collection="research-docs", top_k=5)

    synthesis = await anthropic_client.messages.create(
        model="claude-3-5-sonnet-20241022",
        messages=[{"role": "user", "content": build_synthesis_prompt(plan, context)}],
    )
    total_tokens += synthesis.usage.input_tokens + synthesis.usage.output_tokens

    return WorkflowResult(answer=synthesis.content, tokens=total_tokens)

Tracking total_tokens across every step — not just the final one — is what makes accurate per-workflow billing possible later.

Validating tool calls before executing them

If a step uses Claude's tool-calling to decide on an action (call an API, query a database), never execute the tool call blindly. Validate the arguments against a schema first — a malformed or unexpected tool call executed directly is a real security and correctness risk.

from pydantic import ValidationError

async def execute_tool_call(tool_call: dict):
    tool_name = tool_call["name"]
    schema = TOOL_SCHEMAS.get(tool_name)
    if not schema:
        raise ValueError(f"Unknown tool: {tool_name}")

    try:
        validated_args = schema(**tool_call["input"])
    except ValidationError as e:
        # Return the validation error to the model so it can retry with corrected args
        return {"error": str(e), "retry": True}

    return await TOOL_HANDLERS[tool_name](validated_args)

Returning the validation error back to the model (rather than just failing the request) lets Claude self-correct on the next turn — this materially reduces the number of workflows that fail outright on a single malformed call.

Retrying without duplicating side effects

A network timeout after Claude has already executed a tool call server-side (e.g., sent an email) is worse than the timeout itself if your retry logic re-runs the whole step. Separate "get a response from Claude" retries from "execute a side-effecting action" — only the former should be retried freely.

async def get_claude_response_with_retry(messages: list[dict], max_retries: int = 3):
    for attempt in range(max_retries):
        try:
            return await anthropic_client.messages.create(
                model="claude-3-5-sonnet-20241022",
                messages=messages,
            )
        except anthropic.RateLimitError:
            await asyncio.sleep(2 ** attempt)
    raise WorkflowTimeoutError("Claude did not respond after retries")

# Side-effecting tool executions are NOT wrapped in this retry loop

Setting a hard budget per workflow run

Long chains can run away in cost if a loop condition doesn't terminate cleanly (a common failure mode in agent-style workflows that decide their own next step). Enforce a hard cap independent of the model's own judgment:

MAX_STEPS = 8
MAX_TOKENS_PER_WORKFLOW = 50_000

async def run_bounded_workflow(query: str):
    total_tokens = 0
    for step_num in range(MAX_STEPS):
        if total_tokens > MAX_TOKENS_PER_WORKFLOW:
            return WorkflowResult(status="budget_exceeded", partial=True)
        response = await get_claude_response_with_retry(build_next_step(query, step_num))
        total_tokens += response.usage.input_tokens + response.usage.output_tokens
        if is_final_answer(response):
            return WorkflowResult(status="complete", answer=response.content, tokens=total_tokens)
    return WorkflowResult(status="max_steps_reached", partial=True)

Falling back to OpenAI on provider outage

Since Claude and GPT-4o are both accessible behind a compatible chat interface, a workflow-critical path can fail over to a secondary provider rather than failing the whole request during an outage:

async def chat_with_fallback(messages: list[dict]):
    try:
        return await llm.chat(messages, provider="anthropic")
    except ProviderUnavailableError:
        return await llm.chat(messages, provider="openai")

This only works cleanly if your prompts aren't tightly coupled to one model's specific quirks — worth testing your prompts against both providers periodically, not just at fallback time.

What FastAPI AI Kit includes

The kit's unified LLM layer normalizes token usage and streaming across Anthropic and OpenAI, so the fallback pattern above works with zero provider-specific branching in your business logic. Automatic retry with backoff is built into every provider call, and token tracking aggregates cleanly across multi-step chains for accurate per-workflow billing.