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.

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()

run() does not return on its own. It polls until you call shutdown(), then returns once shutdown is complete. A Worker is also an async context manager: async with worker: starts it on entry and shuts it down on exit, which is how a Worker process usually stops. See Shut down a Worker.

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​

Shut a Worker down by leaving the async with block, which calls shutdown() for you. To keep the Worker running until the process is interrupted, create an asyncio.Event and wait on it inside the block.

Set that event from the entry point that catches KeyboardInterrupt:

interrupt_event = asyncio.Event()

if __name__ == "__main__":
    loop = asyncio.new_event_loop()
    try:
        loop.run_until_complete(main())
    except KeyboardInterrupt:
        interrupt_event.set()
        loop.run_until_complete(loop.shutdown_asyncgens())

Waiting on interrupt_event inside the block holds the Worker open. Once the event is set, the block exits, and 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.