# Run a Worker - Python SDK > For the complete documentation index, see [llms.txt](https://docs.temporal.io/llms.txt). > Any documentation page is available as raw Markdown by appending `.md` to its URL. > 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](/develop/python/client/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](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() ``` `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. See [Shut down a Worker](#shut-down-a-worker). ## Register Workflows and Activities All Workers polling the same Task Queue must register the same Workflow Types and Activity Types. A Task Queue does not route by type, so any Worker polling it can receive any Task on that queue. A Worker that receives a Task for a type it did not register fails that Task. 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. [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, ), ) ``` 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`. [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() ``` See [Worker shutdown](/encyclopedia/workers/worker-shutdown) for what happens to in-flight Workflow Tasks and Activities.