get_work_queues
Retrieve work queues from Prefect with filtering options by name, paused status, and pagination controls to manage workflow automation resources.
Instructions
Get a list of work queues with optional filtering.
Args: limit: Maximum number of work queues to return offset: Number of work queues to skip name: Filter by name is_paused: Filter by paused status
Returns: A list of work queues with their details
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| is_paused | No | ||
| limit | No | ||
| name | No | ||
| offset | No |
Implementation Reference
- src/mcp_prefect/work_queue.py:14-61 (handler)Implementation of the get_work_queues tool handler function using Prefect client to fetch work queues with optional filters.
@mcp.tool async def get_work_queues( limit: Optional[int] = None, offset: Optional[int] = None, name: Optional[str] = None, is_paused: Optional[bool] = None, work_pool_name: Optional[str] = None, ) -> List[Union[types.TextContent, types.ImageContent, types.EmbeddedResource]]: """ Get a list of work queues with optional filtering. Args: limit: Maximum number of work queues to return offset: Number of work queues to skip name: Filter by name is_paused: Filter by paused status work_pool_name: Filter by work pool name Returns: A list of work queues with their details """ async with get_client() as client: # Build filter object work_queue_filter = None if name or is_paused is not None: work_queue_filter = WorkQueueFilter() if name: work_queue_filter.name = WorkQueueFilterName(like_=f"%{name}%") # Note: is_paused filter is not directly available in WorkQueueFilter # We'll handle it by filtering the results after the API call work_queues = await client.read_work_queues( work_queue_filter=work_queue_filter, work_pool_name=work_pool_name, limit=limit, offset=offset or 0 ) # Apply is_paused filter manually if needed if is_paused is not None: work_queues = [wq for wq in work_queues if wq.is_paused == is_paused] work_queues_result = { "work_queues": [work_queue.model_dump() for work_queue in work_queues] } return [types.TextContent(type="text", text=str(work_queues_result))] - src/mcp_prefect/main.py:62-64 (registration)Conditional import of the work_queue module in main.py, which triggers registration of tools (including get_work_queues) via @mcp.tool decorators.
if APIType.WORK_QUEUE.value in apis: info("Loading Work Queue API...") from . import work_queue - src/mcp_prefect/server.py:6-6 (helper)Creation of the FastMCP server instance 'mcp' used by @mcp.tool decorators to register tools.
mcp = FastMCP(f"MCP Prefect {__version__}")