Prefect MCP Server

Instructions

Implementation Reference

• src/mcp_prefect/flow_run.py:17-85 (handler)

  The handler function implementing the 'get_flow_runs' tool logic. It queries the Prefect client for flow runs with applied filters, enriches with UI URLs, and returns as text content.

  @mcp.tool
  async def get_flow_runs(
      limit: Optional[int] = None,
      offset: Optional[int] = None,
      flow_name: Optional[str] = None,
      state_type: Optional[str] = None,
      state_name: Optional[str] = None,
      deployment_id: Optional[str] = None,
      tags: Optional[List[str]] = None,
      start_time_before: Optional[str] = None,
      start_time_after: Optional[str] = None,
  ) -> List[Union[types.TextContent, types.ImageContent, types.EmbeddedResource]]:
      """
      Get a list of flow runs with optional filtering.
      
      Args:
          limit: Maximum number of flow runs to return
          offset: Number of flow runs to skip
          flow_name: Filter by flow name
          state_type: Filter by state type (e.g., "RUNNING", "COMPLETED", "FAILED")
          state_name: Filter by state name
          deployment_id: Filter by deployment ID
          tags: Filter by tags
          start_time_before: ISO formatted datetime string
          start_time_after: ISO formatted datetime string
          
      Returns:
          A list of flow runs with their details
      """
      async with get_client() as client:
          # Build filter parameters
          filters = {}
          if flow_name:
              filters["flow_name"] = {"like_": f"%{flow_name}%"}
          if state_type:
              filters["state"] = {"type": {"any_": [state_type.upper()]}}
          if state_name:
              filters["state"] = {"name": {"any_": [state_name]}}
          if deployment_id:
              filters["deployment_id"] = {"eq_": UUID(deployment_id)}
          if tags:
              filters["tags"] = {"all_": tags}
          if start_time_after:
              filters["start_time"] = {"ge_": start_time_after}
          if start_time_before:
              if "start_time" in filters:
                  filters["start_time"]["le_"] = start_time_before
              else:
                  filters["start_time"] = {"le_": start_time_before}
          
          flow_runs = await client.read_flow_runs(
              limit=limit,
              offset=offset,
              **filters
          )
          
          # Add UI links to each flow run
          flow_runs_result = {
              "flow_runs": [
                  {
                      **flow_run.dict(),
                      "ui_url": get_flow_run_url(str(flow_run.id))
                  }
                  for flow_run in flow_runs
              ]
          }
          
          return [types.TextContent(type="text", text=str(flow_runs_result))]

• src/mcp_prefect/main.py:42-44 (registration)

  The import statement in main.py that loads the flow_run module, triggering registration of the 'get_flow_runs' tool via its @mcp.tool decorator.

  if APIType.FLOW_RUN.value in apis:
      info("Loading Flow Run API...")
      from . import flow_run

• src/mcp_prefect/flow_run.py:12-15 (helper)

  Helper function used by the handler to generate UI URLs for flow runs.

  def get_flow_run_url(flow_run_id: str) -> str:
      base_url = PREFECT_API_URL.replace("/api", "")
      return f"{base_url}/flow-runs/{flow_run_id}"

• src/mcp_prefect/flow_run.py:17-18 (registration)

  The @mcp.tool decorator that registers the get_flow_runs function as an MCP tool.

  @mcp.tool
  async def get_flow_runs(