📣 News: Integration between OpenAI Agents SDK and Temporal is now in public preview. Learn more.

Temporal is a distributed, scalable, durable, and highly available orchestration engine used to execute asynchronous, long-running business logic in a scalable and resilient way.

"Temporal Python SDK" is the framework for authoring workflows and activities using the Python programming language.

Also see:

Application Development Guide - Once you've tried our Quick Start, check out our guide on how to use Temporal in your Python applications, including information around Temporal core concepts.
Python Code Samples
API Documentation - Complete Temporal Python SDK Package reference.

In addition to features common across all Temporal SDKs, the Python SDK also has the following interesting features:

Type Safe

This library uses the latest typing and MyPy support with generics to ensure all calls can be typed. For example, starting a workflow with an int parameter when it accepts a str parameter would cause MyPy to fail.

Different Activity Types

The activity worker has been developed to work with async def, threaded, and multiprocess activities. Threaded activities are the initial recommendation, and further guidance can be found in the docs.

Custom asyncio Event Loop

The workflow implementation basically turns async def functions into workflows backed by a distributed, fault-tolerant event loop. This means task management, sleep, cancellation, etc have all been developed to seamlessly integrate with asyncio concepts.

See the blog post introducing the Python SDK for an informal introduction to the features and their implementation.

Contents

Quick Start
Usage
Development

Quick Start

We will guide you through the Temporal basics to create a "hello, world!" script on your machine. It is not intended as one of the ways to use Temporal, but in reality it is very simplified and decidedly not "the only way" to use Temporal. For more information, check out the docs references in "Next Steps" below the quick start.

Installation

Install the temporalio package from PyPI.

These steps can be followed to use with a virtual environment and pip:

Create a virtual environment
Update pip - python -m pip install -U pip
- Needed because older versions of pip may not pick the right wheel
Install Temporal SDK - python -m pip install temporalio

The SDK is now ready for use. To build from source, see "Building" near the end of this documentation.

NOTE: This README is for the current branch and not necessarily what's released on PyPI.

Implementing a Workflow

Create the following in activities.py:

from temporalio import activity

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

Create the following in workflows.py:

from datetime import timedelta
from temporalio import workflow

# Import our activity, passing it through the sandbox
with workflow.unsafe.imports_passed_through():
    from activities import say_hello

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

Create the following in run_worker.py:

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

# Import the activity and workflow from our other files
from activities import say_hello
from workflows import SayHello

async def main():
    # Create client connected to server at the given address
    client = await Client.connect("localhost:7233")

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

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

Assuming you have a Temporal server running on localhost, this will run the worker:

python run_worker.py

Running a Workflow

Create the following script at run_workflow.py:

import asyncio
from temporalio.client import Client

# Import the workflow from the previous code
from workflows import SayHello

async def main():
    # Create client connected to server at the given address
    client = await Client.connect("localhost:7233")

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

    print(f"Result: {result}")

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

Assuming you have run_worker.py running from before, this will run the workflow:

python run_workflow.py

The output will be:

Result: Hello, my-name!

Next Steps

Temporal can be implemented in your code in many different ways, to suit your application's needs. The links below will give you much more information about

temporalio

Description

Quick Start

Installation

Implementing a Workflow

Running a Workflow

Next Steps