Temporal.io Python SDK

1.24.0 · active · verified Sun Mar 29

The Temporal.io Python SDK (current version 1.24.0) provides a framework for authoring durable workflows and activities using the Python programming language. It enables developers to build resilient, scalable, and fault-tolerant applications that can execute long-running business logic reliably. The library maintains a regular release cadence, often with monthly or bi-monthly updates, introducing new features, improvements, and bug fixes.

Warnings

breaking Python 3.9 support was removed in version 1.19.0 as it reached End-of-Life. Users on Python 3.9 must upgrade to Python 3.10 or newer.
Fix: Upgrade Python environment to 3.10 or a later supported version.
breaking In version 1.23.0, fields `workflow_id`, `workflow_namespace`, `workflow_run_id`, and `workflow_type` within `activity.Info` were made optional. Additionally, `converter.BaseWorkflowSerializationContext` was removed.
Fix: Adjust code that manually accesses `activity.Info` fields to handle potential `None` values. Refactor any custom `BaseWorkflowSerializationContext` subclasses.
breaking Version 1.24.0 introduces general availability for Nexus and OpenAI Agents SDK Integration. It also includes new OpenTelemetry integration for OpenAI Agents, which may cause conflicts or require adjustments for existing custom OpenTelemetry setups.
Fix: Review and update OpenTelemetry tracing configurations if using OpenAI Agents with custom instrumentation. Leverage the GA features of Nexus and OpenAI Agents.
gotcha Beginning with version 1.21.0, providing an `api_key` to `Client.connect` automatically enables TLS.
Fix: If you wish to use an `api_key` without TLS, explicitly pass `tls=False` to `Client.connect`.
gotcha The Workflow Sandbox isolates workflow code to ensure determinism. Importing non-standard library or non-`temporalio` modules directly into workflow files can lead to `RestrictedWorkflowAccessError` or non-determinism.
Fix: For modules needed only by activities, import them within the activity function, or use `with workflow.unsafe.imports_passed_through(): import my_module` when importing them into workflow files. For performance, pass through any side-effect-free third-party libraries explicitly.

Install

pip install temporalio Install the latest version

Imports

Client
```
from temporalio.client import Client
```
Used to connect to a Temporal Cluster and interact with workflows (start, signal, query).
Worker
```
from temporalio.worker import Worker
```
Used to host workflow and activity implementations and poll task queues.
workflow
```
from temporalio import workflow
```
Provides decorators like `@workflow.defn` for defining workflows and functions like `workflow.execute_activity` for orchestrating activities.
activity
```
from temporalio import activity
```
Provides decorators like `@activity.defn` for defining activities.
workflow.unsafe.imports_passed_through
```
from temporalio import workflow
with workflow.unsafe.imports_passed_through():
    from my_app import my_activity_module
```
Crucial for importing activity modules into workflow files when those activity modules have non-deterministic or non-standard library imports, to prevent sandbox errors.

Quickstart

This quickstart demonstrates how to define an activity and a workflow, configure and start a Temporal worker to process tasks, and execute a workflow. It connects to a local Temporal server (defaulting to `localhost:7233`) and includes `os.environ.get` for flexible configuration.

import asyncio
from datetime import timedelta
from temporalio.client import Client
from temporalio.worker import Worker
from temporalio import activity, workflow

# Define an activity
@activity.defn
async def say_hello(name: str) -> str:
    return f"Hello, {name}!"

# Define a workflow
@workflow.defn
class GreetingWorkflow:
    @workflow.run
    async def run(self, name: str) -> str:
        return await workflow.execute_activity(
            say_hello,
            name,
            schedule_to_close_timeout=timedelta(seconds=5),
        )

async def main():
    # Connect to Temporal server (default to localhost:7233)
    # Use os.environ.get('TEMPORAL_HOST_PORT', 'localhost:7233') for dynamic connection
    client = await Client.connect(os.environ.get('TEMPORAL_HOST_PORT', 'localhost:7233'))

    # Run a worker
    task_queue_name = "my-task-queue"
    worker = Worker(
        client,
        task_queue=task_queue_name,
        workflows=[GreetingWorkflow],
        activities=[say_hello],
    )
    # Start worker in background
    worker_task = asyncio.create_task(worker.run())
    print(f"Worker started on task queue '{task_queue_name}'...")

    # Start a workflow execution
    result = await client.execute_workflow(
        GreetingWorkflow.run,
        "Temporal",
        id="greeting-workflow-id",
        task_queue=task_queue_name,
    )
    print(f"Workflow result: {result}") # Expected: "Hello, Temporal!"

    # Clean up worker
    worker_task.cancel()
    await worker_task

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

view raw JSON →