Prefect MCP Server

Instructions

Implementation Reference

• src/mcp_prefect/work_queue.py:140-180 (handler)

  The main handler implementation for the MCP tool 'update_work_queue'. It is decorated with @mcp.tool for registration and handles updating Prefect work queue attributes via the Prefect client, returning the updated queue details as text content.

  @mcp.tool
  async def update_work_queue(
      work_queue_id: str,
      name: Optional[str] = None,
      description: Optional[str] = None,
      is_paused: Optional[bool] = None,
      concurrency_limit: Optional[int] = None,
  ) -> List[Union[types.TextContent, types.ImageContent, types.EmbeddedResource]]:
      """
      Update a work queue.
      
      Args:
          work_queue_id: The work queue UUID
          name: New name
          description: New description
          is_paused: New paused status
          concurrency_limit: New concurrency limit
          
      Returns:
          Details of the updated work queue
      """
      async with get_client() as client:
          # Prepare update data
          update_data = {}
          if name is not None:
              update_data["name"] = name
          if description is not None:
              update_data["description"] = description
          if is_paused is not None:
              update_data["is_paused"] = is_paused
          if concurrency_limit is not None:
              update_data["concurrency_limit"] = concurrency_limit
          
          await client.update_work_queue(
              id=UUID(work_queue_id),
              **update_data
          )
          
          # Read the updated work queue to return its details
          updated_work_queue = await client.read_work_queue(UUID(work_queue_id))
          return [types.TextContent(type="text", text=str(updated_work_queue.model_dump()))]