Prefect MCP Server

Instructions

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__}")