Skip to main content

Standalone Activities - Python SDK

SUPPORT, STABILITY, and DEPENDENCY INFO

Temporal Python SDK support for Standalone Activities is at Pre-release.

All APIs are experimental and may be subject to backwards-incompatible changes.

Standalone Activities are Activities that run independently, without being orchestrated by a Workflow. Instead of starting an Activity from within a Workflow Definition, you start a Standalone Activity directly from a Temporal Client.

The way you write the Activity and register it with a Worker is identical to Workflow Activities. The only difference is that you execute a Standalone Activity directly from your Temporal Client.

This page covers the following:

note

This documentation uses source code derived from the Python sample.

Get Started with Standalone Activities​

Prerequisites:

Temporal Cloud

All code samples on this page use ClientConfig.load_client_connect_config() to configure the Temporal Client connection. By default this connects to localhost:7233, but it responds to environment variables and TOML configuration files, so the same code works against a local dev server and Temporal Cloud without changes. See Run Standalone Activities with Temporal Cloud below.

The first step in running a Standalone Activity involves starting a Temporal server:

temporal server start-dev

This command automatically starts the Temporal development server with the Web UI, and creates the default Namespace. It uses an in-memory database, so do not use it for real use cases.

The Temporal Server will now be available for client connections on localhost:7233, and the Temporal Web UI will now be accessible at http://localhost:8233. Standalone Activities are available from the nav bar item located towards the top left of the page:

Standalone Activities Web UI nav bar item

Clone the samples-python repository to follow along:

git clone https://github.com/temporalio/samples-python.git
cd samples-python

Write an Activity Function​

An Activity in the Temporal Python SDK is just a normal function with the @activity.defn decorator. It can optionally be an async def. The way you write a Standalone Activity is identical to how you write an Activity to be orchestrated by a Workflow. In fact, an Activity can be executed both as a Standalone Activity and as a Workflow Activity.

To see this code in a working example, see the Standalone Activity sample.

# my_activity.py
from dataclasses import dataclass

from temporalio import activity


@dataclass
class ComposeGreetingInput:
    greeting: str
    name: str


@activity.defn
async def compose_greeting(input: ComposeGreetingInput) -> str:
    activity.logger.info("Running activity with parameter %s" % input)
    return f"{input.greeting}, {input.name}!"

Run a Worker with the Activity registered​

Running a Worker for Standalone Activities is the same as running a Worker for Workflow Activities — you create a Worker, register the Activity, and run the Worker. The Worker doesn't need to know whether the Activity will be invoked from a Workflow or as a Standalone Activity. See How to run a Worker for more details on Worker setup and configuration options.

To see this code in a working example, see the Standalone Activity sample.

import asyncio

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

from my_activity import compose_greeting


async def main():
    connect_config = ClientConfig.load_client_connect_config()
    client = await Client.connect(**connect_config)
    worker = Worker(
        client,
        task_queue="hello-standalone-activity-task-queue",
        activities=[compose_greeting],
    )
    await worker.run()


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

Execute a Standalone Activity​

Use client.execute_activity() to execute a Standalone Activity. Call this from your application code, not from inside a Workflow Definition. This durably enqueues your Standalone Activity in the Temporal Server, waits for it to be executed on your Worker, and then fetches the result:

import asyncio
from datetime import timedelta

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

from my_activity import ComposeGreetingInput, compose_greeting


async def my_application():
    connect_config = ClientConfig.load_client_connect_config()
    client = await Client.connect(**connect_config)

    result = await client.execute_activity(
        compose_greeting,
        args=[ComposeGreetingInput("Hello", "World")],
        id="my-standalone-activity-id",
        task_queue="hello-standalone-activity-task-queue",
        start_to_close_timeout=timedelta(seconds=10),
    )
    print(f"Activity result: {result}")


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

Start a Standalone Activity without waiting for the result​

Starting a Standalone Activity means sending a request to the Temporal Server to durably enqueue your Activity job, without waiting for it to be executed by your Worker.

Use client.start_activity() to start your Standalone Activity and get a handle:

activity_handle = await client.start_activity(
    compose_greeting,
    args=[ComposeGreetingInput("Hello", "World")],
    id="my-standalone-activity-id",
    task_queue="hello-standalone-activity-task-queue",
    start_to_close_timeout=timedelta(seconds=10),
)

You can also use client.get_activity_handle() to create a handle to a previously started Standalone Activity:

activity_handle = client.get_activity_handle(
    activity_id="my-standalone-activity-id",
    run_id="the-run-id",
)

You can now use the handle to wait for the result, describe, cancel, or terminate the Activity.

Wait for the result of a Standalone Activity​

Under the hood, calling client.execute_activity() is the same as calling client.start_activity() to durably enqueue the Standalone Activity, and then calling await activity_handle.result() to wait for the activity to be executed and fetch the result:

activity_result = await activity_handle.result()

List Standalone Activities​

Use client.list_activities() to list Standalone Activity Executions that match a List Filter query. The result is an async iterator that yields ActivityExecution entries:

import asyncio

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


async def my_application():
    connect_config = ClientConfig.load_client_connect_config()
    client = await Client.connect(**connect_config)

    activities = client.list_activities(
        query="TaskQueue = 'my-task-queue'",
    )

    async for info in activities:
        print(
            f"ActivityID: {info.activity_id}, Type: {info.activity_type}, Status: {info.status}"
        )


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

The query parameter accepts the same List Filter syntax used for Workflow Visibility. For example, "ActivityType = 'MyActivity' AND Status = 'Running'".

Count Standalone Activities​

Use client.count_activities() to count Standalone Activity Executions that match a List Filter query.

import asyncio

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


async def my_application():
    connect_config = ClientConfig.load_client_connect_config()
    client = await Client.connect(**connect_config)

    resp = await client.count_activities(
        query="TaskQueue = 'my-task-queue'",
    )

    print("Total activities:", resp.count)

    for group in resp.groups:
        print(f"Group {group.group_values}: {group.count}")


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

Run Standalone Activities with Temporal Cloud​

The code samples on this page use ClientConfig.load_client_connect_config(), so the same code works against Temporal Cloud — just configure the connection via environment variables or a TOML profile. No code changes are needed.

For full details on connecting to Temporal Cloud, including Namespace creation, certificate generation, and authentication options, see Connect to Temporal Cloud.

Connect with mTLS​

export TEMPORAL_ADDRESS=<your-namespace>.<your-account-id>.tmprl.cloud:7233
export TEMPORAL_NAMESPACE=<your-namespace>.<your-account-id>
export TEMPORAL_TLS_CLIENT_CERT_PATH='path/to/your/client.pem'
export TEMPORAL_TLS_CLIENT_KEY_PATH='path/to/your/client.key'

Connect with an API key​

export TEMPORAL_ADDRESS=<region>.<cloud_provider>.api.temporal.io:7233
export TEMPORAL_NAMESPACE=<your-namespace>.<your-account-id>
export TEMPORAL_API_KEY=<your-api-key>

Then run the Worker and starter code as shown in the earlier sections.