Skip to main content

Run a Worker - Python SDK

View Markdown

This page covers long-lived Workers that you host and run as persistent processes. For Workers that run on serverless compute like AWS Lambda, see Serverless Workers.

Create and run a Worker​

Create a Worker with a Temporal Client, the Task Queue to poll, and the Workflows and Activities it can execute. Call run() to start polling. The Worker runs until the process is interrupted.

features/snippets/worker/worker.py

client = await Client.connect("localhost:7233")

worker = Worker(
    client,
    task_queue="my-task-queue",
    workflows=[HelloWorkflow],
    activities=[some_activity],
)
await worker.run()

A Worker is also an async context manager, so async with worker: runs it for the duration of a block. See Shut down a Worker for the pattern most Worker processes use.

Register Workflows and Activities​

All Workers listening to the same Task Queue must be registered to handle the same Workflow Types and Activity Types. If a Worker polls a Task for a type it does not know about, the Task fails. The Workflow Execution itself does not fail.

Pass a list of Workflows in workflows, a list of Activities in activities, or both.

Activities defined with async def run on the Worker's event loop. Activities defined with a plain def are synchronous and require an executor, so pass one in activity_executor:

worker = Worker(
    client,
    task_queue="my-task-queue",
    workflows=[MyWorkflow],
    activities=[my_sync_activity],
    activity_executor=ThreadPoolExecutor(5),
)

The same executor can be shared across multiple Workers.

Connect to Temporal Cloud​

To run a Worker against Temporal Cloud, configure the Client connection with your Namespace address and authentication credentials. See Connect to Temporal Cloud for setup instructions.

Configure Worker options​

The Worker constructor takes keyword arguments that control concurrency limits, pollers, timeouts, and caching, including max_concurrent_activities, max_concurrent_workflow_tasks, and max_cached_workflows. The defaults work for most cases.

To tune these values against real load, see Worker performance and the Worker tuning reference.

Run a versioned Worker​

Set a Worker Deployment Version and enable versioning in deployment_config, then set a default versioning behavior for the Workflows on the Worker.

features/snippets/worker/worker.py

worker = Worker(
    client,
    task_queue="my-task-queue",
    workflows=[HelloWorkflow],
    activities=[some_activity],
    deployment_config=WorkerDeploymentConfig(
        version=WorkerDeploymentVersion(
            deployment_name="my-app",
            build_id="1.0",
        ),
        use_worker_versioning=True,
        default_versioning_behavior=VersioningBehavior.PINNED,
    ),
)

To set the behavior per Workflow instead of on the Worker, pass versioning_behavior to @workflow.defn. See Worker Versioning for the available versioning behaviors and how new versions roll out.

Shut down a Worker​

Use the Worker as an async context manager and wait on an event that your signal handler sets. When the block exits, the Worker stops polling for new Tasks and waits for in-flight Tasks to finish, up to graceful_shutdown_timeout.

features/snippets/worker/worker.py

worker = Worker(
    client,
    task_queue="my-task-queue",
    workflows=[HelloWorkflow],
    activities=[some_activity],
    graceful_shutdown_timeout=timedelta(seconds=30),
)
async with worker:
    await interrupt_event.wait()

See Worker shutdown for what happens to in-flight Workflow Tasks and Activities.