python-sdk

name	python-sdk
description	This skill should be used when the user asks to 'create a Temporal workflow in Python', 'write a Python activity', 'use temporalio', 'fix Python workflow determinism', 'debug workflow replay', 'Python workflow logging', or mentions 'Temporal Python SDK'. Provides Python-specific patterns, asyncio guidance, sandbox constraints, and observability guidance.
version	0.1.0

Temporal Python SDK Best Practices

Overview

The Temporal Python SDK (temporalio) provides a fully async, type-safe approach to building durable workflows. Python 3.9+ required. Workflows run in a sandbox by default for determinism protection.

Quick Start

activities/greet.py - Activity definitions (separate file for performance):

from temporalio import activity


@activity.defn
def greet(name: str) -> str:
    return f"Hello, {name}!"

workflows/greeting.py - Workflow definition (import activities through sandbox):

from datetime import timedelta

from temporalio import workflow

with workflow.unsafe.imports_passed_through():
    from activities.greet import greet


@workflow.defn
class GreetingWorkflow:
    @workflow.run
    async def run(self, name: str) -> str:
        return await workflow.execute_activity(
            greet, name, schedule_to_close_timeout=timedelta(seconds=30)
        )

worker.py - Long-running Worker process:

import asyncio
import concurrent.futures

from temporalio.client import Client
from temporalio.worker import Worker

from activities.greet import greet
from workflows.greeting import GreetingWorkflow


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

    with concurrent.futures.ThreadPoolExecutor(max_workers=100) as activity_executor:
        worker = Worker(
            client,
            task_queue="greeting-queue",
            workflows=[GreetingWorkflow],
            activities=[greet],
            activity_executor=activity_executor,
        )
        await worker.run()


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

starter.py - Client code to start a Workflow:

import asyncio

from temporalio.client import Client

from workflows.greeting import GreetingWorkflow


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

    result = await client.execute_workflow(
        GreetingWorkflow.run,
        "World",
        id="greeting-workflow",
        task_queue="greeting-queue",
    )
    print(result)


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

Key Concepts

Workflow Definition

• Use @workflow.defn decorator on class
• Use @workflow.run on the entry point method
• Must be async (async def)
• Use @workflow.signal, @workflow.query, @workflow.update for handlers

Activity Definition

• Use @activity.defn decorator
• Can be sync or async functions
• Default to sync activities - safer and easier to debug
• Sync activities need activity_executor (ThreadPoolExecutor)
• Async activities require async-safe libraries throughout (e.g., aiohttp not requests)

See references/sync-vs-async.md for detailed guidance on choosing between sync and async.

Worker Setup

• Connect client, create Worker with workflows and activities
• Call await worker.run() to start the long-running Worker process
• Use ThreadPoolExecutor as activity_executor for sync activities

Why Determinism Matters: History Replay

Temporal achieves durability through history replay. When a Worker restarts or recovers from a failure, it re-executes the Workflow code from the beginning, but instead of re-running Activities, it uses the results stored in the Event History.

This is why Workflow code must be deterministic:

• During replay, the SDK compares Commands generated by your code against the Events in history
• If the sequence differs (non-determinism), the Worker cannot restore state
• Non-determinism causes NondeterminismError and blocks Workflow progress

The Python SDK's sandbox provides automatic protection against many common non-deterministic operations, but understanding replay helps you write correct code.

See references/determinism.md for detailed rules and safe alternatives.

Determinism Rules Summary

Safe alternatives:

• workflow.now() instead of datetime.now()
• workflow.random() instead of random
• asyncio.sleep() works (sandbox handles it)
• workflow.uuid4() for UUIDs
• workflow.wait() instead of asyncio.wait() (deterministic ordering)
• workflow.as_completed() instead of asyncio.as_completed() (deterministic ordering)

Pass-through for libraries:

with workflow.unsafe.imports_passed_through():
    import pydantic

Replay-Aware Logging

Use workflow.logger inside Workflows for replay-safe logging that avoids duplicate log messages during replay:

@workflow.defn
class MyWorkflow:
    @workflow.run
    async def run(self) -> str:
        workflow.logger.info("Starting workflow")  # Only logs on initial execution
        result = await workflow.execute_activity(...)
        workflow.logger.info(f"Activity result: {result}")
        return result

Use activity.logger in activities for context-aware logging with automatic correlation:

@activity.defn
async def my_activity() -> str:
    activity.logger.info("Processing activity")  # Includes activity context
    return "done"

File Organization Best Practice

Keep Workflow definitions in separate files from Activity definitions. The Python SDK sandbox reloads Workflow definition files on every execution for determinism protection. Minimizing file contents improves Worker performance.

my_temporal_app/
├── workflows/
│   └── greeting.py      # Only Workflow classes
├── activities/
│   └── translate.py     # Only Activity functions/classes
├── worker.py            # Worker setup, imports both
└── starter.py           # Client code to start workflows

In the Workflow file, import Activities through the sandbox:

# workflows/greeting.py
from temporalio import workflow

with workflow.unsafe.imports_passed_through():
    from activities.translate import TranslateActivities

Common Pitfalls

1. Using datetime.now() in workflows - Use workflow.now() instead
2. Blocking in async activities - Use sync activities or async-safe libraries only
3. Missing executor for sync activities - Add activity_executor=ThreadPoolExecutor()
4. Forgetting to heartbeat - Long activities need activity.heartbeat()
5. Using gevent - Incompatible with SDK
6. Using print() in workflows - Use workflow.logger instead for replay-safe logging
7. Mixing Workflows and Activities in same file - Causes unnecessary reloads, hurts performance

Additional Resources

Reference Files

• references/determinism.md - Sandbox behavior, safe alternatives, pass-through pattern, history replay
• references/sync-vs-async.md - Sync vs async activities, event loop blocking, executor configuration
• references/error-handling.md - ApplicationError, retry policies, non-retryable errors, idempotency
• references/testing.md - WorkflowEnvironment, time-skipping, activity mocking
• references/patterns.md - Signals, queries, child workflows, saga pattern
• references/observability.md - Logging, metrics, tracing, Search Attributes
• references/advanced-features.md - Continue-as-new, schedules, updates, interceptors
• references/data-handling.md - Data converters, Pydantic, payload encryption
• references/versioning.md - Patching API, workflow type versioning, Worker Versioning