temporalio__skill-temporal-developer/references/python/python.md

Temporal Python SDK Reference

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 Demo of Temporal

Add Dependency on Temporal: In the package management system of the Python project you are working on, add a dependency on temporalio.

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, start_to_close_timeout=timedelta(seconds=30)
        )

worker.py - Worker setup (registers activity and workflow, runs indefinitely and processes tasks):

import asyncio
import concurrent.futures
from temporalio.client import Client
from temporalio.envconfig import ClientConfig
from temporalio.worker import Worker

# Import the activity and workflow from our other files
from activities.greet import greet
from workflows.greeting import GreetingWorkflow

async def main():
    connect_config = ClientConfig.load_client_connect_config()
    connect_config.setdefault("target_host", "localhost:7233")
    client = await Client.connect(**connect_config)

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

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

Start the dev server: Start temporal server start-dev in the background.

Start the worker: Start python worker.py in the background (appropriately adjust command for your project, like uv run python worker.py)

starter.py - Start a workflow execution:

import asyncio
from temporalio.client import Client
from temporalio.envconfig import ClientConfig
import uuid

# Import the workflow from the previous code
from workflows.greeting import GreetingWorkflow

async def main():
    connect_config = ClientConfig.load_client_connect_config()
    connect_config.setdefault("target_host", "localhost:7233")
    client = await Client.connect(**connect_config)

    # Execute a workflow
    result = await client.execute_workflow(GreetingWorkflow.run, "my name", id=str(uuid.uuid4()), task_queue="my-task-queue")

    print(f"Result: {result}")

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

Run the workflow: Run python starter.py (or uv run, etc.). Should output: Result: Hello, my-name!.

Key Concepts

Workflow Definition

• Use @workflow.defn decorator on class
• Put any state initialization logic in the __init__ of your workflow class to guarantee that it happens before signals/updates arrive. If your state initialization logic requires the workflow parameters, then add the @workflow.init decorator and parameters to your __init__.
• 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 sync-vs-async.md for detailed guidance on choosing between sync and async.

Worker Setup

• Load connection settings with ClientConfig.load_client_connect_config(), connect the client, and create a Worker with workflows and activities
• Run the worker
• Activities can specify custom executor

Determinism

Workflow code must be deterministic!. All sources of non-determinism should either use Temporal-provided actions or (primarily) be defined in Activities. Read references/core/determinism.md and references/python/determinism.md to understand more.

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. Non-deterministic code in workflows - Use activities for all non-deterministic and/or fallible code
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, bad structure
8. Forgetting to wait on activity calls - workflow.execute_activity() is async; you must eventually await it (directly or via asyncio.gather() for parallel execution)

Writing Tests

See references/python/testing.md for info on writing tests.

Additional Resources

Reference Files

• references/python/patterns.md - Signals, queries, child workflows, saga pattern, etc.
• references/python/determinism.md - Sandbox behavior, safe alternatives, pass-through pattern, history replay
• references/python/gotchas.md - Python-specific mistakes and anti-patterns
• references/python/error-handling.md - ApplicationError, retry policies, non-retryable errors, idempotency
• references/python/observability.md - Logging, metrics, tracing, Search Attributes
• references/python/testing.md - WorkflowEnvironment, time-skipping, activity mocking
• references/python/sync-vs-async.md - Sync vs async activities, event loop blocking, executor configuration
• references/python/advanced-features.md - Schedules, worker tuning, and more
• references/python/data-handling.md - Data converters, Pydantic, payload encryption
• references/python/external-storage.md - Claim-check pattern for large payloads (S3 driver, custom drivers, codec-server handling, multi-region durability)
• references/python/versioning.md - Patching API, workflow type versioning, Worker Versioning
• references/python/standalone-activities.md - Standalone Activities: run an Activity directly from a Client without a Workflow (Public Preview). Concept overview at references/core/standalone-activities.md.
• references/python/determinism-protection.md - Python sandbox specifics, forbidden operations, pass-through imports
• references/python/ai-patterns.md - LLM integration, Pydantic data converter, AI workflow patterns
• references/python/workflow-streams.md - Public-Preview temporalio.contrib.workflow_streams library: durable, offset-addressed event channel for streaming progress to subscribers.

Python Integrations

For Python-specific third-party integrations (OpenAI Agents SDK, Google ADK, etc.), see references/integrations.md and filter for Python. Reference files live under references/python/integrations/.