General Temporal

Temporal Workflows — Python Patterns

Temporal is a durable execution engine. Every workflow step is recorded as an immutable event; if the worker crashes, Temporal replays history to resume exactly where it left off.

When to Activate

• Structuring a new Temporal workflow and its activities
• Debugging non-determinism errors, replay failures, or signal issues
• Adding retries, timeouts, or error handling to activities
• Managing state across workflow turns without losing it on crash
• Implementing human-in-the-loop or long-running multi-step pipelines
• Writing or wiring a Temporal worker

---

Core Concepts

Event Sourcing / Replay

Temporal records every decision as an immutable event before executing it:

Event 1: WorkflowStarted
Event 2: ActivityScheduled  (fetch_data, url)
Event 3: ActivityCompleted  (fetch_data, url) → "result"
Event 4: SignalReceived      (approve)
Event 5: ActivityScheduled  (process_data, ...)
          ← worker crashes here

On restart, Temporal replays events 1–4. For completed activities it returns the recorded result — no real I/O. Execution resumes at Event 5 for real.

Determinism rule: Workflow code must produce the same decisions on every replay.

• ✅ Call activities for all I/O, random values, current time
• ❌ Never use random, time.time(), datetime.now(), httpx, or file reads in workflow code
• ❌ Never import I/O libraries at module level in workflow files

Workflow vs Activity

| | Workflow | Activity | |---|---|---| | Purpose | Orchestration, decisions, state | Real I/O — HTTP, DB, LLM, file | | I/O allowed | ❌ Must be deterministic | ✅ Unrestricted | | Retried by Temporal | Workflow tasks retry on exception | Yes, via RetryPolicy | | Current time | workflow.now() only | datetime.now() fine |

---

Minimal Workflow

# workflow.py
from datetime import timedelta
from temporalio import workflow
from temporalio.common import RetryPolicy

from activities import fetch_data, process_data  # imported for type reference only

@workflow.defn
class MyWorkflow:

    @workflow.run
    async def run(self, url: str) -> str:
        # All I/O goes through execute_activity — never call directly
        raw = await workflow.execute_activity(
            fetch_data,
            url,
            start_to_close_timeout=timedelta(minutes=2),
            retry_policy=RetryPolicy(maximum_attempts=3),
        )

        result = await workflow.execute_activity(
            process_data,
            raw,
            start_to_close_timeout=timedelta(minutes=5),
        )

        return result

---

Activities

Activities are the only place with real I/O. Keep them focused — one network call or DB operation per activity.

# activities.py
import httpx
from temporalio import activity

@activity.defn
async def fetch_data(url: str) -> str:
    async with httpx.AsyncClient(timeout=30) as client:
        response = await client.get(url)
        response.raise_for_status()   # non-2xx → exception → Temporal retries
    return response.text

@activity.defn
async def process_data(raw: str) -> str:
    # CPU-bound or DB work here
    return raw.strip().upper()

Activity design rules:

• Accept and return JSON-serializable types (str, int, dict, list, Pydantic models)
• Raise exceptions freely — Temporal catches and retries per RetryPolicy
• Make activities idempotent — they may run more than once on retry
• Keep activities short — long-running ones need heartbeats

---

Worker

# run_worker.py
import asyncio
from temporalio.client import Client
from temporalio.worker import Worker

from workflow import MyWorkflow
from activities import fetch_data, process_data

async def main():
    client = await Client.connect("localhost:7233")

    worker = Worker(
        client,
        task_queue="my-task-queue",
        workflows=[MyWorkflow],
        activities=[fetch_data, process_data],
    )

    print("Worker started")
    await worker.run()

if __name__ == "__main__":
    asyncio.run(main())

---

Starting a Workflow

# client.py
import asyncio
from temporalio.client import Client
from workflow import MyWorkflow

async def main():
    client = await Client.connect("localhost:7233")

    # Start and wait for result
    result = await client.execute_workflow(
        MyWorkflow.run,
        "https://example.com/data",
        id="my-workflow-id-001",      # unique per workflow instance
        task_queue="my-task-queue",
    )
    print(result)

    # Start without waiting (fire and forget)
    handle = await client.start_workflow(
        MyWorkflow.run,
        "https://example.com/data",
        id="my-workflow-id-002",
        task_queue="my-task-queue",
    )
    # Later: result = await handle.result()

asyncio.run(main())

---

Retries and Timeouts

from datetime import timedelta
from temporalio.common import RetryPolicy

# Full retry config
result = await workflow.execute_activity(
    fetch_data,
    url,
    # How long one attempt can run
    start_to_close_timeout=timedelta(minutes=2),

    # How long all attempts combined can run
    schedule_to_close_timeout=timedelta(minutes=10),

    retry_policy=RetryPolicy(
        initial_interval=timedelta(seconds=1),   # first retry after 1s
        backoff_coefficient=2.0,                  # doubles each retry
        maximum_interval=timedelta(seconds=30),   # cap at 30s
        maximum_attempts=5,                       # 5 total attempts, then raise
        non_retryable_error_types=["ValueError"], # don't retry these
    ),
)

| Timeout | Scope | Use for | |---|---|---| | start_to_close_timeout | Single attempt | Normal activity duration limit | | schedule_to_close_timeout | All attempts | Hard deadline across all retries | | schedule_to_start_timeout | Queue wait time | Detect stuck workers |

---

Signals and Queries

@workflow.defn
class ApprovalWorkflow:

    def __init__(self):
        self._approved = False
        self._status = "pending"

    @workflow.run
    async def run(self, item_id: str) -> str:
        # Block until approved (or timeout)
        await workflow.wait_condition(
            lambda: self._approved,
            timeout=timedelta(hours=24),   # give up after 24h
        )
        return await workflow.execute_activity(
            process_item, item_id,
            start_to_close_timeout=timedelta(minutes=5),
        )

    @workflow.signal
    async def approve(self) -> None:
        self._approved = True
        self._status = "approved"

    @workflow.signal
    async def reject(self, reason: str) -> None:
        self._status = f"rejected: {reason}"
        raise Exception(f"Rejected: {reason}")

    @workflow.query
    def status(self) -> str:
        return self._status

# Send a signal from a client
handle = client.get_workflow_handle("approval-workflow-id")
await handle.signal(ApprovalWorkflow.approve)

# Query current state without interrupting
status = await handle.query(ApprovalWorkflow.status)

---

State Management

Workflows are stateful by design — instance variables persist across signals and replay.

@workflow.defn
class BatchWorkflow:

    def __init__(self):
        self._results: list[str] = []
        self._errors: list[str] = []

    @workflow.run
    async def run(self, urls: list[str]) -> dict:
        for url in urls:
            try:
                result = await workflow.execute_activity(
                    fetch_data, url,
                    start_to_close_timeout=timedelta(minutes=2),
                    retry_policy=RetryPolicy(maximum_attempts=2),
                )
                self._results.append(result)
            except Exception as e:
                self._errors.append(f"{url}: {e}")

        return {"results": self._results, "errors": self._errors}

For state that must survive worker replacement (long-running workflows across deployments), persist it in an external store (Postgres, Redis) via an activity and reload it on startup.

@workflow.run
async def run(self, workflow_id: str) -> str:
    # Load persisted state at the start of each run
    state = await workflow.execute_activity(
        load_state, workflow_id,
        start_to_close_timeout=timedelta(seconds=10),
    )
    # ... do work, update state via save_state activity ...

---

Long-Running Activities (Heartbeats)

Activities that take longer than start_to_close_timeout must send heartbeats — otherwise Temporal assumes the worker is dead and retries.

@activity.defn
async def process_large_file(file_path: str) -> str:
    lines = open(file_path).readlines()
    results = []

    for i, line in enumerate(lines):
        result = expensive_operation(line)
        results.append(result)

        # Heartbeat every 100 lines — keeps the activity alive
        if i % 100 == 0:
            activity.heartbeat(f"processed {i}/{len(lines)} lines")

    return "\n".join(results)

# In workflow — set heartbeat_timeout shorter than start_to_close_timeout
await workflow.execute_activity(
    process_large_file,
    file_path,
    start_to_close_timeout=timedelta(hours=1),
    heartbeat_timeout=timedelta(seconds=30),   # fail if no heartbeat in 30s
)

---

Child Workflows

from temporalio.workflow import ChildWorkflowHandle

@workflow.defn
class ParentWorkflow:

    @workflow.run
    async def run(self, items: list[str]) -> list[str]:
        # Launch child workflows concurrently
        handles: list[ChildWorkflowHandle] = []
        for item in items:
            handle = await workflow.start_child_workflow(
                ChildWorkflow.run,
                item,
                id=f"child-{item}",
                task_queue="my-task-queue",
            )
            handles.append(handle)

        # Wait for all to complete
        return list(await asyncio.gather(*[h.result() for h in handles]))

---

Testing

# test_workflow.py
import pytest
from temporalio.testing import WorkflowEnvironment
from temporalio.worker import Worker

from workflow import MyWorkflow
from activities import fetch_data, process_data

@pytest.mark.asyncio
async def test_my_workflow():
    async with await WorkflowEnvironment.start_time_skipping() as env:
        async with Worker(
            env.client,
            task_queue="test-queue",
            workflows=[MyWorkflow],
            activities=[fetch_data, process_data],
        ):
            result = await env.client.execute_workflow(
                MyWorkflow.run,
                "https://example.com",
                id="test-workflow-1",
                task_queue="test-queue",
            )
            assert result == "EXPECTED OUTPUT"

# Mock activities for unit testing the workflow logic
from unittest.mock import AsyncMock

@pytest.mark.asyncio
async def test_workflow_with_mocked_activities():
    async with await WorkflowEnvironment.start_time_skipping() as env:
        mock_fetch = AsyncMock(return_value="raw data")
        mock_process = AsyncMock(return_value="processed")

        async with Worker(
            env.client,
            task_queue="test-queue",
            workflows=[MyWorkflow],
            activities=[mock_fetch, mock_process],
        ):
            result = await env.client.execute_workflow(
                MyWorkflow.run, "https://example.com",
                id="test-2", task_queue="test-queue",
            )
            assert result == "processed"

---

Common Errors

| Error | Cause | Fix | |---|---|---| | workflow.NondeterminismError | Workflow code changed after workflows started | Never change the order/type of execute_activity calls; version with workflow.patched() | | ActivityError / ApplicationError | Activity raised after exhausting retries | Catch in workflow, notify user, continue or abort | | Signal dropped | Workflow already completed when signal arrived | Send signals before the workflow finishes, or use update instead of signal | | schedule_to_start_timeout exceeded | No workers polling the task queue | Start a worker on the same task queue | | Activity runs twice | Worker crashed after activity completed but before Temporal recorded it | Make activities idempotent |

---

Versioning (Safe Code Changes)

# Use workflow.patched() to change workflow logic without breaking running workflows
@workflow.run
async def run(self, url: str) -> str:
    if workflow.patched("use-v2-processor"):
        # New code path — for workflows started after this deploy
        result = await workflow.execute_activity(
            process_data_v2, url,
            start_to_close_timeout=timedelta(minutes=5),
        )
    else:
        # Old code path — for workflows already in flight
        result = await workflow.execute_activity(
            process_data, url,
            start_to_close_timeout=timedelta(minutes=5),
        )
    return result

Once all pre-patch workflows complete, remove the else branch and the patched() call.

---

Red Flags

• I/O directly in workflow code — httpx, database queries, or open() calls in a workflow function break determinism; on replay the call fires again and may return a different result, causing NondeterminismError; all I/O must live in activities
• random, time.time(), or datetime.now() in a workflow — these return different values on every replay; use workflow.now() for timestamps and route all randomness through activity return values
• Activities that are not idempotent — Temporal may run an activity more than once (crash between execution and recording); an activity that charges a card or sends an email twice on retry is dangerous; use idempotency keys or check-before-act patterns
• Missing start_to_close_timeout — omitting a timeout lets a hung activity block the workflow forever; always set both start_to_close_timeout and a RetryPolicy
• Long-running activities without heartbeats — Temporal assumes a silent activity is dead after heartbeat_timeout; any activity that runs longer than a few minutes must call activity.heartbeat() periodically
• Changing activity call order after workflows are in flight — adding, removing, or reordering execute_activity calls in a running workflow causes NondeterminismError on replay; use workflow.patched() to safely introduce new code paths
• Using asyncio.create_task inside a workflow — spawning raw tasks in workflow code bypasses Temporal's scheduler and breaks determinism; use child workflows or signals for concurrent branching

Checklist

• [ ] All HTTP, DB, and I/O calls are in activities — zero I/O in workflow functions
• [ ] No random, time.time(), datetime.now(), or I/O imports at module level in workflow files
• [ ] Every execute_activity call has start_to_close_timeout and RetryPolicy
• [ ] Activities are idempotent — safe to run more than once
• [ ] Long-running activities call activity.heartbeat() and have heartbeat_timeout set
• [ ] workflow.execute_activity wrapped in try/except to handle exhausted retries gracefully
• [ ] Workflow ID is unique and deterministic per business entity (e.g. f"order-{order_id}")
• [ ] Code changes to running workflows use workflow.patched() for safe versioning
• [ ] Tests use WorkflowEnvironment.start_time_skipping() to run timers instantly
• [ ] Worker registers all activity functions and workflow classes on the correct task queue

Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

• Author: kid-sid
• Source: kid-sid/claude-spellbook
• License: MIT

Install and usage instructions live in the source repository linked above.