Prefect MCP Server

Instructions

Implementation Reference

• src/mcp_prefect/task_run.py:154-205 (handler)

  The core handler function decorated with @mcp.tool, which registers and implements the get_task_runs_by_flow_run tool. It retrieves task runs associated with a specific flow run using the Prefect client, supports filtering by state_type, limit, and offset, adds UI links, and returns formatted results.

  @mcp.tool
  async def get_task_runs_by_flow_run(
      flow_run_id: str,
      limit: Optional[int] = None,
      offset: Optional[int] = None,
      state_type: Optional[str] = None,
  ) -> List[Union[types.TextContent, types.ImageContent, types.EmbeddedResource]]:
      """
      Get task runs for a specific flow run.
      
      Args:
          flow_run_id: The flow run UUID
          limit: Maximum number of task runs to return
          offset: Number of task runs to skip
          state_type: Filter by state type (e.g., "RUNNING", "COMPLETED", "FAILED")
          
      Returns:
          A list of task runs for the specified flow run
      """
      async with get_client() as client:
          # Build filter using new filter objects
          flow_run_filter = FlowRunFilter(
              id=FlowRunFilterId(any_=[UUID(flow_run_id)])
          )
          
          task_run_filter = None
          if state_type:
              task_run_filter = TaskRunFilter(
                  state=TaskRunFilterState(
                      type=TaskRunFilterStateType(any_=[state_type.upper()])
                  )
              )
          
          task_runs = await client.read_task_runs(
              flow_run_filter=flow_run_filter,
              task_run_filter=task_run_filter,
              limit=limit,
              offset=offset or 0
          )
          
          # Add UI links to each task run
          task_runs_result = {
              "task_runs": [
                  {
                      **task_run.model_dump(),
                      "ui_url": get_task_run_url(str(task_run.id))
                  }
                  for task_run in task_runs
              ]
          }
          
          return [types.TextContent(type="text", text=str(task_runs_result))]