temporal-workflows

name	temporal-workflows
description	Expert guidance for building durable, fault-tolerant applications with Temporal. Use when implementing long-running workflows, retry logic, activity orchestration, state persistence, failure recovery, or distributed transactions. Covers Temporal Python SDK, workflow patterns, and production best practices.

Temporal Workflow Orchestration Skill

Overview

Temporal is a durable execution platform that enables building scalable, fault-tolerant applications. Workflows automatically survive process crashes, infrastructure failures, and can resume from exactly where they left off.

When to Use This Skill

• Building long-running business processes that must complete reliably
• Orchestrating multi-step operations with automatic retry
• Implementing saga patterns and distributed transactions
• Creating scheduled/cron workflows
• Building applications that need to survive failures gracefully
• Coordinating multiple services or microservices

Core Concepts

1. Temporal Server Setup

# Using Docker (development)
docker run -d --name temporal-server \
  -p 7233:7233 -p 8080:8080 \
  temporalio/auto-setup:latest

# Using Temporal CLI (alternative)
brew install temporal
temporal server start-dev

Access Temporal Web UI at: http://localhost:8080

2. Workflows

Workflows are the core orchestration unit - durable, resumable, and deterministic.

from datetime import timedelta
from dataclasses import dataclass
from temporalio import workflow

with workflow.unsafe.imports_passed_through():
    from activities import process_document, extract_entities, generate_report

@dataclass
class ProcessingInput:
    document_path: str
    options: dict

@dataclass
class ProcessingResult:
    entities: list[dict]
    report_url: str
    processing_time: float

@workflow.defn(name="DocumentProcessingWorkflow")
class DocumentProcessingWorkflow:
    """Orchestrates document processing with automatic failure recovery."""
    
    @workflow.run
    async def run(self, input: ProcessingInput) -> ProcessingResult:
        # Step 1: Process document
        processed = await workflow.execute_activity(
            process_document,
            input.document_path,
            start_to_close_timeout=timedelta(minutes=5),
            retry_policy=workflow.RetryPolicy(
                initial_interval=timedelta(seconds=1),
                maximum_interval=timedelta(minutes=1),
                backoff_coefficient=2.0,
                maximum_attempts=3,
            ),
        )
        
        # Step 2: Extract entities
        entities = await workflow.execute_activity(
            extract_entities,
            processed,
            start_to_close_timeout=timedelta(minutes=10),
        )
        
        # Step 3: Generate report
        report_url = await workflow.execute_activity(
            generate_report,
            entities,
            start_to_close_timeout=timedelta(minutes=2),
        )
        
        return ProcessingResult(
            entities=entities,
            report_url=report_url,
            processing_time=workflow.now().timestamp(),
        )

3. Activities

Activities are the building blocks that perform actual work (API calls, I/O, etc.).

from temporalio import activity
from dataclasses import dataclass

@dataclass
class DocumentInput:
    path: str
    options: dict

@activity.defn(name="process_document")
async def process_document(doc_path: str) -> dict:
    """Activity that processes a document. Can make network calls, use I/O."""
    # Heartbeat for long-running activities
    activity.heartbeat("Processing started")
    
    # Your actual processing logic
    result = await load_and_process(doc_path)
    
    activity.heartbeat("Processing complete")
    return result

@activity.defn(name="extract_entities")
async def extract_entities(processed_doc: dict) -> list[dict]:
    """Extract entities using LLM - can fail and retry."""
    try:
        return await call_llm_api(processed_doc)
    except RateLimitError:
        # Temporal will retry based on retry policy
        raise

4. Workers

Workers execute workflows and activities.

import asyncio
from temporalio.client import Client
from temporalio.worker import Worker
from workflows import DocumentProcessingWorkflow
from activities import process_document, extract_entities, generate_report

async def main():
    # Connect to Temporal server
    client = await Client.connect("localhost:7233")
    
    # Create worker
    worker = Worker(
        client,
        task_queue="document-processing",
        workflows=[DocumentProcessingWorkflow],
        activities=[process_document, extract_entities, generate_report],
    )
    
    # Run worker (blocks until shutdown)
    await worker.run()

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

5. Client (Starting Workflows)

from temporalio.client import Client
from workflows import DocumentProcessingWorkflow, ProcessingInput

async def start_workflow():
    client = await Client.connect("localhost:7233")
    
    # Start workflow execution
    handle = await client.start_workflow(
        DocumentProcessingWorkflow.run,
        ProcessingInput(document_path="/docs/contract.pdf", options={}),
        id="doc-processing-12345",  # Unique workflow ID
        task_queue="document-processing",
    )
    
    # Wait for result
    result = await handle.result()
    print(f"Workflow completed: {result}")
    
    # Or execute (start + wait) in one call
    result = await client.execute_workflow(
        DocumentProcessingWorkflow.run,
        ProcessingInput(document_path="/docs/contract.pdf", options={}),
        id="doc-processing-12346",
        task_queue="document-processing",
    )

Retry Policies

Configure automatic retry behavior for activities:

from temporalio.common import RetryPolicy
from datetime import timedelta

# Default retry policy
default_retry = RetryPolicy(
    initial_interval=timedelta(seconds=1),
    backoff_coefficient=2.0,
    maximum_interval=timedelta(minutes=5),
    maximum_attempts=10,
)

# Non-retryable errors
api_retry = RetryPolicy(
    initial_interval=timedelta(seconds=2),
    maximum_attempts=5,
    non_retryable_error_types=["ValueError", "ValidationError"],  # Don't retry these
)

# In workflow
result = await workflow.execute_activity(
    my_activity,
    args,
    start_to_close_timeout=timedelta(minutes=5),
    retry_policy=api_retry,
)

Activity Timeouts

await workflow.execute_activity(
    my_activity,
    args,
    # How long the activity can run
    start_to_close_timeout=timedelta(minutes=10),
    
    # How long to wait in queue before starting
    schedule_to_start_timeout=timedelta(minutes=5),
    
    # Total time from schedule to completion
    schedule_to_close_timeout=timedelta(minutes=15),
    
    # For long activities - heartbeat interval
    heartbeat_timeout=timedelta(seconds=30),
)

Workflow Patterns

Pattern 1: Parallel Activity Execution

@workflow.run
async def run(self, items: list[str]) -> list[dict]:
    # Start all activities in parallel
    handles = [
        workflow.start_activity(
            process_item,
            item,
            start_to_close_timeout=timedelta(minutes=5),
        )
        for item in items
    ]
    
    # Wait for all to complete
    results = await asyncio.gather(*handles)
    return results

Pattern 2: Saga with Compensation

@workflow.run
async def run(self, order: Order) -> str:
    compensations = []
    
    try:
        # Step 1: Reserve inventory
        await workflow.execute_activity(reserve_inventory, order)
        compensations.append((release_inventory, order))
        
        # Step 2: Charge payment
        await workflow.execute_activity(charge_payment, order)
        compensations.append((refund_payment, order))
        
        # Step 3: Ship order
        await workflow.execute_activity(ship_order, order)
        
        return "Order completed"
        
    except Exception as e:
        # Compensate in reverse order
        for compensation_activity, args in reversed(compensations):
            await workflow.execute_activity(
                compensation_activity,
                args,
                start_to_close_timeout=timedelta(minutes=5),
            )
        raise

Pattern 3: Signals and Queries

@workflow.defn
class OrderWorkflow:
    def __init__(self):
        self.status = "pending"
        self.items = []
    
    @workflow.signal
    async def add_item(self, item: str):
        """Signal: modify workflow state from outside."""
        self.items.append(item)
    
    @workflow.signal
    async def complete_order(self):
        self.status = "completed"
    
    @workflow.query
    def get_status(self) -> dict:
        """Query: read workflow state without modifying."""
        return {"status": self.status, "items": self.items}
    
    @workflow.run
    async def run(self) -> str:
        # Wait for completion signal
        await workflow.wait_condition(lambda: self.status == "completed")
        return f"Order completed with {len(self.items)} items"

Pattern 4: Child Workflows

@workflow.run
async def run(self, batch: list[str]) -> list[dict]:
    # Start child workflows for each item
    results = []
    for item in batch:
        result = await workflow.execute_child_workflow(
            ItemProcessingWorkflow.run,
            item,
            id=f"item-{item}",
        )
        results.append(result)
    return results

Workflow Determinism Rules

Workflows MUST be deterministic. Avoid:

# ❌ BAD - Non-deterministic
import random
import datetime

@workflow.run
async def run(self):
    value = random.randint(1, 100)  # Different on replay!
    current_time = datetime.datetime.now()  # Different on replay!
    await asyncio.sleep(5)  # Wrong - use workflow.sleep!

# ✅ GOOD - Deterministic
@workflow.run
async def run(self):
    value = await workflow.execute_activity(get_random_value, ...)
    current_time = workflow.now()  # Temporal's deterministic time
    await workflow.sleep(timedelta(seconds=5))  # Correct!

Allowed in Workflows:

• workflow.execute_activity()
• workflow.sleep() / workflow.wait_condition()
• workflow.now() for current time
• workflow.info() for workflow metadata
• Pure Python logic (no I/O)

NOT Allowed in Workflows:

• Network calls, file I/O, database queries
• random, uuid.uuid4(), datetime.now()
• time.sleep(), asyncio.sleep()
• Global mutable state

Production Best Practices

1. Use dataclasses for inputs/outputs - Type-safe and serializable
2. Set appropriate timeouts - Always specify activity timeouts
3. Implement heartbeats - For long-running activities
4. Use retry policies - Configure based on failure modes
5. Make activities idempotent - They may be retried
6. Use workflow IDs - Prevent duplicate executions
7. Monitor with Web UI - http://localhost:8080

Installation

pip install temporalio>=1.9.0

References

• Temporal Documentation
• Python SDK Guide
• Python API Reference
• Python Samples
• GitHub Repository