# Run a Worker - Python SDK

> Create and run a Temporal Worker using the Python SDK.

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](/develop/python/workers/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.

<!--SNIPSTART python-create-worker-->
[features/snippets/worker/worker.py](https://github.com/temporalio/features/blob/main/features/snippets/worker/worker.py)
```py
client = await Client.connect("localhost:7233")

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

`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](#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`:

```python
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](/develop/python/client/temporal-client#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](/develop/worker-performance) and the [Worker tuning reference](/develop/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.

<!--SNIPSTART python-versioned-worker-->
[features/snippets/worker/worker.py](https://github.com/temporalio/features/blob/main/features/snippets/worker/worker.py)
```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,
    ),
)
```
<!--SNIPEND-->

To set the behavior per Workflow instead of on the Worker, pass `versioning_behavior` to `@workflow.defn`.
See [Worker Versioning](/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`:

```python
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`.

<!--SNIPSTART python-worker-graceful-shutdown-->
[features/snippets/worker/worker.py](https://github.com/temporalio/features/blob/main/features/snippets/worker/worker.py)
```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()
```
<!--SNIPEND-->

See [Worker shutdown](/encyclopedia/workers/worker-shutdown) for what happens to in-flight Workflow Tasks and Activities.