> ## Documentation Index > Fetch the complete documentation index at: https://docs.prefect.io/llms.txt > Use this file to discover all available pages before exploring further. # workers # `prefect.server.api.workers` Routes for interacting with work queue objects. ## Functions ### `create_work_pool` ```python theme={null} create_work_pool(work_pool: schemas.actions.WorkPoolCreate, db: PrefectDBInterface = Depends(provide_database_interface), prefect_client_version: Optional[str] = Depends(dependencies.get_prefect_client_version)) -> schemas.responses.WorkPoolResponse ``` Creates a new work pool. If a work pool with the same name already exists, an error will be raised. For more information, see [https://docs.prefect.io/v3/concepts/work-pools](https://docs.prefect.io/v3/concepts/work-pools). ### `read_work_pool` ```python theme={null} read_work_pool(work_pool_name: str = Path(..., description='The work pool name', alias='name'), worker_lookups: WorkerLookups = Depends(WorkerLookups), db: PrefectDBInterface = Depends(provide_database_interface), prefect_client_version: Optional[str] = Depends(dependencies.get_prefect_client_version)) -> schemas.responses.WorkPoolResponse ``` Read a work pool by name ### `read_work_pools` ```python theme={null} read_work_pools(work_pools: Optional[schemas.filters.WorkPoolFilter] = None, limit: int = dependencies.LimitBody(), offset: int = Body(0, ge=0), db: PrefectDBInterface = Depends(provide_database_interface), prefect_client_version: Optional[str] = Depends(dependencies.get_prefect_client_version)) -> List[schemas.responses.WorkPoolResponse] ``` Read multiple work pools ### `count_work_pools` ```python theme={null} count_work_pools(work_pools: Optional[schemas.filters.WorkPoolFilter] = Body(None, embed=True), db: PrefectDBInterface = Depends(provide_database_interface)) -> int ``` Count work pools ### `update_work_pool` ```python theme={null} update_work_pool(work_pool: schemas.actions.WorkPoolUpdate, work_pool_name: str = Path(..., description='The work pool name', alias='name'), worker_lookups: WorkerLookups = Depends(WorkerLookups), db: PrefectDBInterface = Depends(provide_database_interface)) -> None ``` Update a work pool ### `delete_work_pool` ```python theme={null} delete_work_pool(work_pool_name: str = Path(..., description='The work pool name', alias='name'), worker_lookups: WorkerLookups = Depends(WorkerLookups), db: PrefectDBInterface = Depends(provide_database_interface)) -> None ``` Delete a work pool ### `read_work_pool_concurrency_status` ```python theme={null} read_work_pool_concurrency_status(work_pool_name: str = Path(..., description='The work pool name', alias='name'), page: int = Body(1, ge=1), limit: int = dependencies.LimitBody(), flow_run_limit: int = Body(10, ge=0, le=200, description='Max flow runs per queue'), worker_lookups: WorkerLookups = Depends(WorkerLookups), db: PrefectDBInterface = Depends(provide_database_interface)) -> schemas.responses.WorkPoolConcurrencyStatus ``` Read concurrency status for a work pool, including per-queue breakdown with flow run summaries. Queues are paginated; flow runs per queue are capped by flow\_run\_limit. ### `get_scheduled_flow_runs` ```python theme={null} get_scheduled_flow_runs(docket: dependencies.Docket, work_pool_name: str = Path(..., description='The work pool name', alias='name'), work_queue_names: List[str] = Body(None, description='The names of work pool queues'), scheduled_before: DateTime = Body(None, description='The maximum time to look for scheduled flow runs'), scheduled_after: DateTime = Body(None, description='The minimum time to look for scheduled flow runs'), limit: int = dependencies.LimitBody(), worker_lookups: WorkerLookups = Depends(WorkerLookups), db: PrefectDBInterface = Depends(provide_database_interface)) -> List[schemas.responses.WorkerFlowRunResponse] ``` Load scheduled runs for a worker ### `create_work_queue` ```python theme={null} create_work_queue(work_queue: schemas.actions.WorkQueueCreate, work_pool_name: str = Path(..., description='The work pool name'), worker_lookups: WorkerLookups = Depends(WorkerLookups), db: PrefectDBInterface = Depends(provide_database_interface)) -> schemas.responses.WorkQueueResponse ``` Creates a new work pool queue. If a work pool queue with the same name already exists, an error will be raised. For more information, see [https://docs.prefect.io/v3/concepts/work-pools#work-queues](https://docs.prefect.io/v3/concepts/work-pools#work-queues). ### `read_work_queue` ```python theme={null} read_work_queue(work_pool_name: str = Path(..., description='The work pool name'), work_queue_name: str = Path(..., description='The work pool queue name', alias='name'), worker_lookups: WorkerLookups = Depends(WorkerLookups), db: PrefectDBInterface = Depends(provide_database_interface)) -> schemas.responses.WorkQueueResponse ``` Read a work pool queue ### `read_work_queues` ```python theme={null} read_work_queues(work_pool_name: str = Path(..., description='The work pool name'), work_queues: schemas.filters.WorkQueueFilter = None, limit: int = dependencies.LimitBody(), offset: int = Body(0, ge=0), worker_lookups: WorkerLookups = Depends(WorkerLookups), db: PrefectDBInterface = Depends(provide_database_interface)) -> List[schemas.responses.WorkQueueResponse] ``` Read all work pool queues ### `update_work_queue` ```python theme={null} update_work_queue(work_queue: schemas.actions.WorkQueueUpdate, work_pool_name: str = Path(..., description='The work pool name'), work_queue_name: str = Path(..., description='The work pool queue name', alias='name'), worker_lookups: WorkerLookups = Depends(WorkerLookups), db: PrefectDBInterface = Depends(provide_database_interface)) -> None ``` Update a work pool queue ### `delete_work_queue` ```python theme={null} delete_work_queue(work_pool_name: str = Path(..., description='The work pool name'), work_queue_name: str = Path(..., description='The work pool queue name', alias='name'), worker_lookups: WorkerLookups = Depends(WorkerLookups), db: PrefectDBInterface = Depends(provide_database_interface)) -> None ``` Delete a work pool queue ### `worker_channel_connect` ```python theme={null} worker_channel_connect(websocket: WebSocket, work_pool_name: str = Path(..., description='The work pool name'), db: PrefectDBInterface = Depends(provide_database_interface)) -> None ``` ### `worker_heartbeat` ```python theme={null} worker_heartbeat(work_pool_name: str = Path(..., description='The work pool name'), name: str = Body(..., description='The worker process name', embed=True), heartbeat_interval_seconds: Optional[int] = Body(None, description="The worker's heartbeat interval in seconds", embed=True), worker_lookups: WorkerLookups = Depends(WorkerLookups), db: PrefectDBInterface = Depends(provide_database_interface)) -> None ``` ### `read_workers` ```python theme={null} read_workers(work_pool_name: str = Path(..., description='The work pool name'), workers: Optional[schemas.filters.WorkerFilter] = None, limit: int = dependencies.LimitBody(), offset: int = Body(0, ge=0), worker_lookups: WorkerLookups = Depends(WorkerLookups), db: PrefectDBInterface = Depends(provide_database_interface)) -> List[schemas.responses.WorkerResponse] ``` Read all worker processes ### `delete_worker` ```python theme={null} delete_worker(work_pool_name: str = Path(..., description='The work pool name'), worker_name: str = Path(..., description="The work pool's worker name", alias='name'), worker_lookups: WorkerLookups = Depends(WorkerLookups), db: PrefectDBInterface = Depends(provide_database_interface)) -> None ``` Delete a work pool's worker ## Classes ### `WorkerLookups` ### `WorkerChannelSetupError` ### `WorkerChannelWorkPoolUpdateEvent`