Building Fault-Tolerant Long-Running AI Workflows with Claude Agent SDK × Temporal.io — A Production Design Guide to Durable Execution and Saga Patterns

The Core Architecture: Claude Agent SDK × Temporal

The guiding principle when combining these two is simple: Workflows must not call Claude API directly; wrap every call in an Activity. Follow this single rule and Temporal takes care of retries, persistence, and observability.

# Recommended structure
# ┌──────────── Workflow ─────────────┐
# │ agent_loop_workflow(input) {       │
# │   while not done:                   │
# │     response = await act.call_claude(messages)  # Activity
# │     if response.is_tool_use:        │
# │       result = await act.run_tool(tool_name, args)  # Activity
# │       messages.append(result)       │
# │     else:                           │
# │       done = True                   │
# │   return final_answer               │
# │ }                                   │
# └─────────────────────────────────────┘

The payoff is that when act.call_claude() fails due to timeout or rate limiting, Temporal retries it with exponential backoff automatically. There is no try/except or retry decorator in the Workflow code. And if the worker dies, when it restarts, the Event History tells it exactly where to resume.

Project Setup

Here are the dependencies. Python 3.11 or newer is assumed.

# Install dependencies
pip install temporalio anthropic python-dotenv
 
# Project structure
# workflows/
#   ├── __init__.py
#   ├── agent_workflow.py      # Workflow definitions
#   ├── activities.py          # Activity definitions (Claude API calls)
#   ├── worker.py              # Worker entry point
#   └── starter.py             # Client to start workflows

Activity: A Production-Ready Claude API Call

Activities are where we wrap Claude API calls. The critical design decisions here are idempotency and heartbeating.

# activities.py
import os
from anthropic import Anthropic, APIError, RateLimitError
from temporalio import activity
from dataclasses import dataclass
from typing import Any
 
client = Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])
 
@dataclass
class ClaudeCallInput:
    messages: list[dict[str, Any]]
    system: str
    tools: list[dict[str, Any]]
    max_tokens: int = 4096
    model: str = "claude-sonnet-4-5"
 
@dataclass
class ClaudeCallResult:
    stop_reason: str
    content: list[dict[str, Any]]
    usage_input_tokens: int
    usage_output_tokens: int
 
@activity.defn
async def call_claude(inp: ClaudeCallInput) -> ClaudeCallResult:
    """
    Claude API call as an Activity. Temporal handles retries for timeouts
    and rate limits, so the right behavior here is to let relevant exceptions
    propagate rather than swallowing them.
    """
    # Heartbeat: tell the worker and server "I'm still alive" during long calls
    activity.heartbeat("calling claude")
 
    try:
        response = client.messages.create(
            model=inp.model,
            max_tokens=inp.max_tokens,
            system=inp.system,
            tools=inp.tools,
            messages=inp.messages,
        )
    except RateLimitError as e:
        # Let retriable errors propagate; Temporal's RetryPolicy handles backoff
        raise
    except APIError as e:
        if e.status_code and e.status_code >= 500:
            # 5xx is transient; let it retry
            raise
        # 4xx is a bad request — retrying won't help
        raise activity.ApplicationError(
            f"Claude API client error: {e.status_code} {e}",
            non_retryable=True,
        )
 
    return ClaudeCallResult(
        stop_reason=response.stop_reason,
        content=[block.model_dump() for block in response.content],
        usage_input_tokens=response.usage.input_tokens,
        usage_output_tokens=response.usage.output_tokens,
    )

The key decision inside an Activity is to cleanly separate retriable from non-retriable errors. Without non_retryable=True, Temporal will retry indefinitely. Authentication errors and malformed requests (4xx) cannot be fixed by retrying, so we stop them explicitly.

Idempotent Tool Execution

Tool calls that touch a database or an external API need careful idempotency handling. Temporal guarantees Activities run at least once, but retries after a timeout might cause two executions.

# activities.py (continued)
import hashlib
from typing import Any
 
@dataclass
class ToolRunInput:
    tool_name: str
    tool_input: dict[str, Any]
    idempotency_key: str  # Generated on the Workflow side
 
@activity.defn
async def run_tool(inp: ToolRunInput) -> dict[str, Any]:
    """
    Executes a tool the agent has invoked. Operations with side effects
    use idempotency_key to guard against duplicate execution.
    """
    if inp.tool_name == "send_invoice":
        return await _send_invoice_idempotent(inp.tool_input, inp.idempotency_key)
    elif inp.tool_name == "fetch_customer":
        # Idempotent read operations run directly
        return await _fetch_customer(inp.tool_input)
    else:
        raise activity.ApplicationError(
            f"Unknown tool: {inp.tool_name}",
            non_retryable=True,
        )
 
async def _send_invoice_idempotent(args: dict, idem_key: str) -> dict:
    """
    Sends an invoice through Stripe using its Idempotency-Key header.
    If the same key arrives twice, Stripe returns the original response.
    """
    # In real code, pass idempotency_key to stripe.Invoice.create()
    return {"invoice_id": f"inv_{idem_key[:16]}", "status": "sent"}

The Workflow generates the idempotency key and hands it to the Activity. Because Workflows are deterministic, re-executions produce the same key, guaranteeing that retried Activities see the same identifier.

The Workflow: The Agent Loop

The Workflow assembles the conversation loop with Claude. This is the architectural heart. It looks almost identical to ordinary agent code, except every API call flows through an Activity.

# agent_workflow.py
from datetime import timedelta
from temporalio import workflow
from temporalio.common import RetryPolicy
 
# Avoid circular imports of types
with workflow.unsafe.imports_passed_through():
    from .activities import (
        call_claude, run_tool,
        ClaudeCallInput, ClaudeCallResult, ToolRunInput,
    )
 
@workflow.defn
class InvoiceProcessingWorkflow:
    def __init__(self) -> None:
        self._messages: list[dict] = []
        self._human_approval: bool | None = None
 
    @workflow.run
    async def run(self, customer_ids: list[str]) -> dict:
        """
        A long-running workflow that processes invoices for a list of customers.
        Expected to take hours. Pauses mid-flow for human approval.
        """
        system = "You are an invoice processing agent. Issue invoices for the customers indicated."
        tools = [
            {
                "name": "fetch_customer",
                "description": "Fetch customer info",
                "input_schema": {"type": "object", "properties": {"customer_id": {"type": "string"}}},
            },
            {
                "name": "send_invoice",
                "description": "Send an invoice",
                "input_schema": {
                    "type": "object",
                    "properties": {
                        "customer_id": {"type": "string"},
                        "amount": {"type": "number"},
                    },
                },
            },
        ]
 
        self._messages = [
            {"role": "user", "content": f"Issue invoices for customers {customer_ids}."}
        ]
 
        # Retry policy: up to 5 attempts with exponential backoff
        retry_policy = RetryPolicy(
            initial_interval=timedelta(seconds=1),
            maximum_interval=timedelta(minutes=5),
            maximum_attempts=5,
            non_retryable_error_types=["ApplicationError"],
        )
 
        results = []
        # The agent loop
        for _ in range(50):  # safety cap on iterations
            claude_result = await workflow.execute_activity(
                call_claude,
                ClaudeCallInput(
                    messages=self._messages,
                    system=system,
                    tools=tools,
                ),
                start_to_close_timeout=timedelta(minutes=2),
                retry_policy=retry_policy,
            )
 
            self._messages.append({"role": "assistant", "content": claude_result.content})
 
            if claude_result.stop_reason == "end_turn":
                break
 
            if claude_result.stop_reason == "tool_use":
                tool_results = []
                for block in claude_result.content:
                    if block["type"] != "tool_use":
                        continue
 
                    # Deterministic idempotency key derived from workflow ID + tool call
                    idem_key = workflow.info().workflow_id + "_" + block["id"]
 
                    # Wait for human approval before send_invoice
                    if block["name"] == "send_invoice":
                        await workflow.wait_condition(
                            lambda: self._human_approval is not None,
                            timeout=timedelta(hours=24),
                        )
                        if not self._human_approval:
                            tool_results.append({
                                "type": "tool_result",
                                "tool_use_id": block["id"],
                                "content": "Approval denied. Skipping.",
                            })
                            self._human_approval = None
                            continue
                        self._human_approval = None  # Reset for the next approval gate
 
                    result = await workflow.execute_activity(
                        run_tool,
                        ToolRunInput(
                            tool_name=block["name"],
                            tool_input=block["input"],
                            idempotency_key=idem_key,
                        ),
                        start_to_close_timeout=timedelta(minutes=1),
                        retry_policy=retry_policy,
                    )
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block["id"],
                        "content": str(result),
                    })
                    results.append(result)
 
                self._messages.append({"role": "user", "content": tool_results})
 
        return {
            "customers_processed": customer_ids,
            "results": results,
            "total_messages": len(self._messages),
        }
 
    @workflow.signal
    def submit_approval(self, approved: bool) -> None:
        """Receive a human approval via Signal from the outside."""
        self._human_approval = approved
 
    @workflow.query
    def get_progress(self) -> dict:
        """Expose progress via Query for monitoring."""
        return {
            "messages_count": len(self._messages),
            "waiting_for_approval": self._human_approval is None,
        }

Three things deserve emphasis in this code.

First, every workflow.execute_activity call is recorded in Temporal's Event History. If a worker dies, another worker replays history to reconstruct state. Both requests and responses to Claude API are persisted, so re-execution continues the same conversation.

Second, workflow.wait_condition "pauses" the Workflow but releases worker memory. Temporal remembers "this workflow is waiting for an approval Signal," and when the Signal arrives it wakes the workflow up. A human approver can take four hours without anything running in memory on your side.

Third, the retry policy is declarative. 5xx errors retry up to five times with exponential backoff automatically; only errors matching non_retryable_error_types propagate to the Workflow. Error handling lives inside Activities, so the Workflow reads like a plain loop.

Saga Compensation for Partial Failures

In production, you always hit the case where something fails mid-workflow and you need to roll back side effects you have already committed. Temporal lets you write explicit Saga patterns to compose compensation safely.

# Adds to agent_workflow.py
from contextlib import suppress
 
@workflow.defn
class RefundableInvoiceWorkflow:
    @workflow.run
    async def run(self, customer_id: str, amount: float) -> dict:
        compensations = []  # Stack of rollback actions
 
        try:
            # Step 1: Check credit
            credit_result = await workflow.execute_activity(
                run_tool,
                ToolRunInput(
                    tool_name="check_credit",
                    tool_input={"customer_id": customer_id},
                    idempotency_key=f"{workflow.info().workflow_id}_credit",
                ),
                start_to_close_timeout=timedelta(seconds=30),
            )
 
            # Step 2: Reserve inventory
            reservation = await workflow.execute_activity(
                run_tool,
                ToolRunInput(
                    tool_name="reserve_inventory",
                    tool_input={"customer_id": customer_id, "amount": amount},
                    idempotency_key=f"{workflow.info().workflow_id}_reserve",
                ),
                start_to_close_timeout=timedelta(seconds=30),
            )
            compensations.append(("release_inventory", reservation))
 
            # Step 3: Issue invoice (if this fails, roll back Step 2)
            invoice = await workflow.execute_activity(
                run_tool,
                ToolRunInput(
                    tool_name="send_invoice",
                    tool_input={"customer_id": customer_id, "amount": amount},
                    idempotency_key=f"{workflow.info().workflow_id}_invoice",
                ),
                start_to_close_timeout=timedelta(minutes=1),
            )
            compensations.append(("void_invoice", invoice))
 
            return {"status": "success", "invoice": invoice}
 
        except Exception as e:
            # Run compensations in reverse order (the Saga principle)
            for tool_name, state in reversed(compensations):
                with suppress(Exception):
                    await workflow.execute_activity(
                        run_tool,
                        ToolRunInput(
                            tool_name=tool_name,
                            tool_input=state,
                            idempotency_key=f"{workflow.info().workflow_id}_comp_{tool_name}",
                        ),
                        start_to_close_timeout=timedelta(seconds=30),
                    )
            raise

Compensation actions are themselves retriable Activities. Since compensation can also fail, we wrap individual compensations in with suppress(Exception) so one failure does not halt the rest of the rollback. That decision is business-policy dependent: "alert and continue if compensation fails" versus "treat a failed compensation as critical and hand off to a recovery workflow" — decide up front.

Five Production Pitfalls

This is the most valuable section of the article. These are the non-obvious traps I hit while putting Claude Agent SDK + Temporal into production.

Pitfall 1: Calling datetime.now() or random Inside a Workflow

Workflows must be deterministic, so non-deterministic code breaks replay consistency with the Event History.

# BAD: produces different values on replay and breaks history reconstruction
import datetime
timestamp = datetime.datetime.now()
 
# GOOD: use Temporal's deterministic APIs
timestamp = workflow.now()
random_val = workflow.random().randint(0, 100)

Pitfall 2: Claude Responses Exploding Your Event History

Whatever Claude returns is stored verbatim in the Event History. By default, Event History is capped around 2 MB; exceed it and the Workflow is terminated.

# BAD: pass a huge document through the Activity
@activity.defn
async def summarize_document(full_text: str) -> str:
    # If full_text is 500 KB, the next turn pushes 500 KB + summary into the history
    result = await call_claude(...)
    return full_text + "\n\nSummary: " + result
 
# GOOD: keep large payloads in external storage; pass references
@activity.defn
async def summarize_document(doc_id: str) -> str:
    full_text = await s3_get(doc_id)
    summary = await call_claude(...)
    summary_id = await s3_put(summary)
    return summary_id  # Only the ID lands in Event History

The rule in production is: if a Claude response is likely to exceed 10 KB, stash it in S3 or DynamoDB. I underestimated this early on and got my workflows terminated for history size on day two.

Pitfall 3: Only Setting start_to_close_timeout

Temporal has four timeout flavors, each with distinct semantics.

• schedule_to_start_timeout: Time from being queued until a worker picks the Activity up
• start_to_close_timeout: Time the Activity is allowed to run once a worker has it
• schedule_to_close_timeout: Total end-to-end time including queuing
• heartbeat_timeout: Maximum allowed gap between heartbeats

Claude API call duration is unpredictable, so the practical pattern is a generous start_to_close_timeout combined with a heartbeat_timeout to distinguish "making progress" from "hanging."

# Configuration we use in production
await workflow.execute_activity(
    call_claude,
    inp,
    start_to_close_timeout=timedelta(minutes=10),  # leeway for long generations
    heartbeat_timeout=timedelta(seconds=30),       # detect hangs quickly
    retry_policy=retry_policy,
)

Pitfall 4: Forgetting Continue-As-New on Long Workflows

A single Workflow execution's Event History grows over time, increasing the cost of state replay. A good rule of thumb: if you exceed 50,000 events, use Continue-As-New.

@workflow.defn
class LongRunningAgent:
    @workflow.run
    async def run(self, state: dict) -> dict:
        iteration = state.get("iteration", 0)
 
        while iteration < 1000000:
            # ...agent work...
            iteration += 1
 
            # Every 1,000 iterations, Continue-As-New to reset the history
            if iteration % 1000 == 0:
                workflow.continue_as_new({"iteration": iteration, "snapshot": self._state})

Continue-As-New ends the current Workflow execution and starts a new one under the same Workflow ID. Event History resets, keeping memory and persistence costs constant.

Pitfall 5: Unbounded wait_condition

Always set a timeout on workflow.wait_condition. Waiting forever for human approval guarantees the workflow turns into a zombie the moment your ops team forgets about it.

# BAD: blocks forever if no Signal arrives
await workflow.wait_condition(lambda: self._human_approval is not None)
 
# GOOD: explicit timeout with a fallback path
try:
    await workflow.wait_condition(
        lambda: self._human_approval is not None,
        timeout=timedelta(hours=48),
    )
except asyncio.TimeoutError:
    # Default deny, alert, escalate — pick based on business rules
    self._human_approval = False
    await workflow.execute_activity(
        notify_timeout, inp, start_to_close_timeout=timedelta(seconds=30),
    )

Child Workflows for Agent Hierarchies

For larger work, having a parent Workflow fan out into multiple Child Workflows running in parallel is a useful pattern. Each Child has its own Event History, keeping the parent's size in check.

@workflow.defn
class OrchestratorWorkflow:
    @workflow.run
    async def run(self, customer_batches: list[list[str]]) -> list[dict]:
        """Process customer batches in parallel as independent Child Workflows."""
        handles = []
        for batch_idx, batch in enumerate(customer_batches):
            handle = await workflow.start_child_workflow(
                InvoiceProcessingWorkflow.run,
                batch,
                id=f"{workflow.info().workflow_id}_batch_{batch_idx}",
                retry_policy=RetryPolicy(maximum_attempts=3),
            )
            handles.append(handle)
 
        # Await all children
        results = []
        for handle in handles:
            try:
                result = await handle
                results.append(result)
            except Exception as e:
                # Tolerate per-batch failure; log and continue
                results.append({"error": str(e)})
        return results

The parent only orchestrates startup and wait, so even 100 parallel batches leave its Event History slim.

Integrating Streaming Claude Responses

So far our Activities have called client.messages.create() synchronously and returned once Claude finishes. That works for tool-use workflows where responses are usually a few paragraphs at most. For user-facing workflows that need streaming — think a live support agent that speaks to a human while the Workflow orchestrates downstream actions — you need a different shape.

The trick is to keep the Workflow boundary at the message level, not the token level. The Activity streams internally and only returns once the full message is assembled. For real-time UX, a separate path (a WebSocket server alongside the Worker) handles token-level streaming to the client, while the Workflow only sees finalized messages. This is the split I see most often in production: Temporal governs the business logic, and a thin streaming proxy handles the end-user experience.

@activity.defn
async def call_claude_streaming(inp: ClaudeCallInput) -> ClaudeCallResult:
    """Collect a streamed response, pushing deltas to a side channel as they arrive."""
    collected_content = []
    input_tokens = 0
    output_tokens = 0
 
    # Heartbeat inside the async iterator keeps Temporal aware of progress
    with client.messages.stream(
        model=inp.model,
        max_tokens=inp.max_tokens,
        system=inp.system,
        tools=inp.tools,
        messages=inp.messages,
    ) as stream:
        for event in stream:
            activity.heartbeat(f"stream event: {event.type}")
            # Push token deltas to Redis pub/sub for the WebSocket proxy
            if event.type == "content_block_delta":
                await redis.publish(f"ws:{inp.session_id}", event.delta.text)
 
        final_message = stream.get_final_message()
        collected_content = [block.model_dump() for block in final_message.content]
        input_tokens = final_message.usage.input_tokens
        output_tokens = final_message.usage.output_tokens
 
    return ClaudeCallResult(
        stop_reason=final_message.stop_reason,
        content=collected_content,
        usage_input_tokens=input_tokens,
        usage_output_tokens=output_tokens,
    )

Two details matter here. First, the activity.heartbeat inside the stream loop prevents Temporal from timing out the Activity during a very long generation. Second, the Workflow still receives a single ClaudeCallResult once the message is complete, which keeps Event History clean. If you tried to stream individual tokens through the Workflow, you would flood the Event History with thousands of events per message.

Handling Long Tool Executions

Some tools an AI agent invokes are themselves long-running. A code-generation agent might call run_pytest and that test suite might take twelve minutes. Wrapping such a tool directly in an Activity means one start_to_close_timeout has to cover a range from "unit test" (seconds) to "full e2e suite" (many minutes). Heartbeating helps, but there is a better pattern: split the tool into "start" and "poll for result" Activities, with the Workflow checking in periodically.

@activity.defn
async def start_test_run(args: dict, idem_key: str) -> str:
    """Kick off a test run asynchronously and return a job ID."""
    job_id = await test_runner.submit(args["branch"], idem_key)
    return job_id
 
@activity.defn
async def poll_test_status(job_id: str) -> dict:
    """Return current status without blocking."""
    return await test_runner.status(job_id)
 
# In the Workflow
job_id = await workflow.execute_activity(start_test_run, ..., start_to_close_timeout=timedelta(seconds=30))
 
while True:
    await workflow.sleep(timedelta(seconds=15))
    status = await workflow.execute_activity(
        poll_test_status, job_id, start_to_close_timeout=timedelta(seconds=10),
    )
    if status["state"] in ("success", "failed"):
        break

Using workflow.sleep is important: it is a durable sleep. The Workflow can wait for hours without consuming Worker memory. Compare this to asyncio.sleep, which would keep the Worker tied up. I have had a single Worker managing thousands of workflows sleeping simultaneously without meaningful memory pressure, because workflow.sleep is handled entirely by the Temporal server.

Production Monitoring and Observability

The Temporal Web UI is great, but it is not enough on its own for production. You want these signals wired into external monitoring.

Workflow start, complete, and failure counts: Pull Temporal's metrics into Prometheus and visualize them in Grafana.

Activity failure rates and retry counts: A sudden rise in retries on one Activity often hints at downstream service degradation.

Claude API cost aggregation: Include usage_input_tokens / usage_output_tokens in Activity return values, then write them to a cost table at completion. Keying by Workflow ID lets you answer "how much did this particular workflow run cost?"

# Add a cost-recording Activity to activities.py
@activity.defn
async def record_claude_cost(workflow_id: str, input_tokens: int, output_tokens: int) -> None:
    """Record cost after each Claude call"""
    # Example Sonnet 4.5 pricing: $3/Mtok input, $15/Mtok output
    cost_usd = (input_tokens * 3.0 + output_tokens * 15.0) / 1_000_000
    await db_insert_cost(workflow_id, cost_usd, input_tokens, output_tokens)

For deeper observability, Claude API × OpenTelemetry: A Complete Guide to AI Application Observability pairs nicely and shows concretely how to instrument the Activity layer with OpenTelemetry. Claude API Production Resilience Patterns complements this article on general production design.

When Plain Agent Code Is Still the Right Choice

Before adopting Temporal, it helps to name the cases where it is the wrong tool. If your entire workflow finishes in under a minute, latency matters more than durability, and a simple retry with exponential backoff inside a single process is enough. A chat endpoint that hits Claude once per request is a terrible fit for Temporal — the extra round-trip to the Temporal server per Activity adds latency you cannot justify.

Batch jobs that run once a night and have no human interaction are another marginal case. A well-written cron job with idempotent steps and alerting is often simpler than a Temporal deployment. The question I ask is: "If this process gets interrupted at the 80% mark, what is the cost of just re-running the whole thing?" When the answer is "we waste a few dollars and thirty minutes," skip the complexity. When the answer is "we double-charge customers or lose irreplaceable state," Temporal starts to pay for itself immediately.

The inflection point I observed in my own projects was around the moment a workflow crossed four distinct failure modes: Claude rate limits, downstream API timeouts, human approval delays, and partial batch failures. When any three of those live inside the same process, hand-rolling resilience becomes a second product you are maintaining alongside your actual product. That is when Temporal's declarative retries and event history stop feeling like overhead and start feeling like relief.

Versioning Workflows Without Breaking In-Flight Executions

Long-running workflows create a problem short-lived code does not: by the time a workflow completes, you may have deployed three new versions of the code. Temporal's workflow.patched API lets you introduce changes without breaking executions that started under the old code.

@workflow.defn
class UpgradableWorkflow:
    @workflow.run
    async def run(self, inp: dict) -> dict:
        # Old code path: simple fetch
        if workflow.patched("use_parallel_fetch"):
            # New code path: parallel fetches across regions
            results = await asyncio.gather(*[
                workflow.execute_activity(
                    fetch_from_region, region,
                    start_to_close_timeout=timedelta(seconds=30),
                )
                for region in ["us-east", "eu-west", "ap-northeast"]
            ])
        else:
            # Legacy path kept until all in-flight workflows have drained
            results = [await workflow.execute_activity(
                fetch_legacy, inp,
                start_to_close_timeout=timedelta(minutes=1),
            )]
 
        return {"results": results}

Workflows that started before the patch continue on the old path; new workflows take the new one. After you are confident no old-version workflows remain, you can replace workflow.patched with workflow.deprecate_patch and eventually remove the legacy branch. This small discipline saves you from incidents where a deployment silently breaks workflows that have been running for two days.

Real Numbers from Production

On one of my personal projects, running a Claude Agent SDK workflow that processes roughly 400 multi-step tasks per day, moving to Temporal produced measurable changes. Mean time to recovery from worker crashes dropped from about twelve minutes (manual resubmission after grepping logs) to effectively zero (the workflow resumed automatically on the surviving worker). The rate of duplicate-execution incidents fell from two or three per month — usually the result of someone manually retrying a job that had already partially completed — to zero, because the idempotency key flow made duplicate tool invocations impossible.

The less glamorous but more valuable change was psychological. I stopped worrying about deploying during the day. Previously, any deployment risked interrupting a running workflow and leaving it in a half-committed state. Now, a deployment kills the worker, Temporal hands the workflow to the next worker to come online, and execution continues from the last committed event. This alone has changed how often I ship small fixes, because the cost of deploying dropped to nearly nothing.

Cost itself moved in an interesting direction. Temporal's per-workflow overhead is real — you pay for the server (or for Temporal Cloud's per-action pricing). For my workload, the direct infrastructure bill rose slightly. But the hidden costs it eliminated were larger: engineering time spent on retry logic, on-call time spent reconstructing state from logs, and the dollar cost of duplicate Claude API calls from manual restarts. Net, I spend less total money and substantially less attention on this class of problem.

Testing Strategy: Using Temporal TestEnvironment

Temporal's temporalio.testing.WorkflowEnvironment lets you test Workflow logic without a running server.

# test_invoice_workflow.py
import pytest
from temporalio.testing import WorkflowEnvironment
from temporalio.worker import Worker
from workflows.agent_workflow import InvoiceProcessingWorkflow
from workflows.activities import call_claude, run_tool
 
@pytest.mark.asyncio
async def test_invoice_workflow_with_mock_claude():
    async with await WorkflowEnvironment.start_time_skipping() as env:
        # Swap the Activity for a mock
        async def mock_call_claude(inp):
            return ClaudeCallResult(
                stop_reason="end_turn",
                content=[{"type": "text", "text": "Complete"}],
                usage_input_tokens=100,
                usage_output_tokens=50,
            )
 
        async with Worker(
            env.client,
            task_queue="test-queue",
            workflows=[InvoiceProcessingWorkflow],
            activities=[mock_call_claude, run_tool],
        ):
            result = await env.client.execute_workflow(
                InvoiceProcessingWorkflow.run,
                ["customer_1"],
                id="test-workflow",
                task_queue="test-queue",
            )
            assert result["customers_processed"] == ["customer_1"]

start_time_skipping() can fast-forward time, so timeout behavior and "what happens 24 hours later" can be exercised in milliseconds. Being able to test the whole workflow without making real Claude API calls is a significant win.

For a broader view of testing, The Complete Guide to Testing Strategies for Claude API Applications covers unit, integration, and E2E design in depth.

A Worked Example: Multi-Day Document Review Workflow

To make the abstract patterns concrete, let's walk through a workflow I built recently: a contract review system where a Claude agent reads a PDF, extracts key clauses, flags risks, and waits for a lawyer's approval before finalizing. The whole cycle takes anywhere from two hours to three days.

The Workflow starts when a user uploads a contract. The first Activity extracts text from the PDF (stored in S3 by reference, as we discussed earlier). The second calls Claude with a system prompt tuned for legal analysis, asking it to produce a structured risk report. The Workflow then enters a wait state using workflow.wait_condition, expecting a Signal from the lawyer review panel within 72 hours. During that wait, the Workflow consumes essentially no resources — Temporal simply holds its state.

When the lawyer submits their verdict (approve, reject, or request revisions) through a Slack button that posts to our API and sends a Signal, the Workflow wakes up. If approved, a final Claude call generates the executive summary for the customer and an email Activity sends it. If revisions are requested, the Workflow loops back to the analysis phase with the lawyer's comments added to the context. If rejected, a compensating Activity notifies the customer with a polite decline template.

What made this production-ready was not any single feature but the combination: Event History captures every lawyer comment and every Claude response, so audit trails are free. The 72-hour wait_condition timeout defaults to "auto-reject and escalate" if no lawyer responds, preventing zombie workflows. workflow.patched let us add a second Claude pass for contracts over a certain length without breaking workflows already in flight. Child Workflows handle multi-contract batches — a customer uploading a bundle of documents spawns one parent and N children, so individual failures do not take down the whole batch.

The surprising win was visibility for non-engineers. Our legal ops team uses the Temporal Web UI directly to see "which contracts are waiting on which lawyer." They do not need a custom dashboard. The built-in filters on Workflow ID, type, and status give them what they need. Two months in, the product manager stopped asking me for weekly status reports because she could pull the view herself.

Common Migration Mistakes from Plain Agent Loops

When migrating existing Claude Agent SDK code to Temporal, I see a few recurring missteps worth naming.

Moving too much into the Workflow. Engineers new to Temporal sometimes lift their whole agent loop into @workflow.defn and keep calling Claude directly. The Workflow then becomes non-deterministic and breaks on replay. The mental model fix: Workflows are pure, Activities are effects. When in doubt, the code belongs in an Activity.

Making Activities too granular. On the other extreme, wrapping every dictionary access in an Activity creates massive Event History overhead. The guideline is: an Activity is a unit of work with side effects that should retry as a whole. A Claude API call is one Activity. Fetching a customer record is one Activity. But calling dict.get("foo") is not.

Ignoring the deterministic constraint on imports. Importing modules that do I/O at import time — reading environment variables, connecting to Redis — will trigger determinism violations. Put such imports inside the with workflow.unsafe.imports_passed_through(): block as shown earlier, or move them into Activity files.

Not budgeting for Temporal's learning curve. The first workflow takes longer to write than the equivalent agent loop. Expect a week of ramp-up. The second workflow is faster. By the fifth, you are thinking in Temporal idioms and moving faster than you did with hand-rolled retries, because you are not debugging retry logic anymore. Give the team room for this learning curve when planning.

Deploying and Scaling Workers

Workers typically live in a container platform such as Kubernetes or ECS. Here is a straightforward worker bootstrap.

# worker.py
import asyncio
import os
from temporalio.client import Client
from temporalio.worker import Worker
from workflows.agent_workflow import InvoiceProcessingWorkflow, RefundableInvoiceWorkflow
from workflows.activities import call_claude, run_tool, record_claude_cost
 
async def main():
    client = await Client.connect(
        os.environ.get("TEMPORAL_ADDRESS", "localhost:7233"),
        namespace=os.environ.get("TEMPORAL_NAMESPACE", "default"),
    )
 
    worker = Worker(
        client,
        task_queue="claude-agent-queue",
        workflows=[InvoiceProcessingWorkflow, RefundableInvoiceWorkflow],
        activities=[call_claude, run_tool, record_claude_cost],
        max_concurrent_activities=20,
        max_concurrent_workflow_tasks=50,
    )
    await worker.run()
 
if __name__ == "__main__":
    asyncio.run(main())

Scaling is straightforward: add more Workers horizontally. Connect additional Workers to the same Task Queue and Temporal distributes work automatically. In practice, tune max_concurrent_activities against Claude API rate limits.

Designing for Multi-Model Workflows

Real production systems rarely use a single Claude model for everything. You route simple classification to Haiku for cost, reasoning-heavy steps to Sonnet, and the genuinely hard synthesis work to Opus. Temporal lets you encode these routing decisions clearly because the model choice lives in the ClaudeCallInput passed to each Activity, and the Workflow controls which input goes where.

The pattern I have settled on is a thin routing helper that the Workflow calls before invoking the Claude Activity. It looks at the task type and message history to pick a model. Each route becomes a separate Activity signature, which means retry policies and timeouts can differ per model. A Haiku call that should take a second gets a 20-second start_to_close_timeout and retries aggressively; an Opus call that might take ninety seconds gets five minutes and retries more conservatively because each retry costs meaningfully more.

What this approach buys you is a clean migration path when models change. Anthropic's model lineup evolves; when a new version is released, you update the routing helper, deploy the new Worker, and in-flight workflows pick up the change at their next Claude call via workflow.patched. You do not need to rewrite agent logic to take advantage of model improvements — you swap the model string behind a single abstraction.

If you are managing per-model cost and latency carefully, the companion article on intelligent model routing in this publication is worth reading alongside. The routing logic described there maps cleanly onto a set of Temporal Activities, one per tier.

The Mental Shift That Takes Longest

The hardest adjustment when adopting Temporal is not the syntax. It is the shift from "my code runs once" to "my code runs every time Temporal needs to rebuild state." Workflow code is replayed. Whenever you add a print() that you expect to see once per execution, you see it many times during replay in the logs — once for each rebuild from history.

This becomes intuitive after a week or two, but early on it feels unsettling. The concrete habit change is: put all your logging inside Activities. Anything printed from Workflow code will appear multiple times in logs and will likely mislead your debugging. Anything printed from Activity code prints exactly once per Activity attempt, which is what you want.

The second habit is trusting the Event History. When something looks wrong, open the Temporal Web UI, navigate to the workflow execution, and read the events. The event stream tells you exactly what Claude was called with and what it returned, in order. I used to reach for log aggregators first; now I reach for the Event History first and only consult logs for Activity-internal details. This changes the feel of debugging from archaeological excavation to reading a well-organized receipt.

What Actually Changed After Adopting This

Three things noticeably shifted once this stack was in production.

First, incident response time dropped. Previously, I would grep logs or query SQL to answer "how far did we get?" Now, the Temporal Web UI visualizes every step by Workflow ID. The retry button on a failed Activity is a genuine relief.

Second, adding human approvals became nearly free. workflow.wait_condition plus a Signal is enough to implement multi-hour or multi-day approvals safely. Wire a Slack button to send the Signal and the whole thing starts to look like a real business system.

Third, cost tracking got easy. Aggregating Claude API consumption by Workflow ID lets you immediately answer "how much did this particular customer cost to process?" Combined with the techniques in Claude API Cost Optimization Production Patterns, you typically find another round of reductions.

Further Reading

Temporal's design philosophy is similar in spirit to AWS Step Functions and Azure Durable Functions, but it is self-hostable, supports many languages, and has a lightweight local development story. Two recommended deep-dives: Designing Data-Intensive Applications (Martin Kleppmann) covers the event-sourcing and consistency foundations that underlie Durable Execution; Building Microservices, 2nd Edition (Sam Newman) has excellent chapters on Saga patterns and event-driven design that complement the compensation material in this article.

One Concrete First Step

When you finish this article, run temporal server start-dev to spin up a local environment. Then pick the longest-running process in your existing Claude API project and port only that one to a Temporal Workflow. Once you feel what it is like to kill the worker, restart it, and watch the workflow pick up where it left off, you will not go back.

As a decision rule: if a workflow takes more than thirty minutes, or involves human approval, or coordinates multiple services, Temporal earns its keep. For simpler, sub-thirty-minute tasks, plain Claude Agent SDK is enough. You do not need to put everything on Temporal. Knowing where it pays off — and where it does not — is the real craft behind a long-lived architecture.