python-sdk - SKILL.md Agent Skill

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

Using datetime.now() in workflows - Use workflow.now() instead
Blocking in async activities - Use sync activities or async-safe libraries only
Missing executor for sync activities - Add activity_executor=ThreadPoolExecutor()
Forgetting to heartbeat - Long activities need activity.heartbeat()
Using gevent - Incompatible with SDK
Using print() in workflows - Use workflow.logger instead for replay-safe logging
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