# 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. The Worker runs until the process is interrupted.

<!--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-->

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

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

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

<!--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.