get_task_runs

Instructions

Implementation Reference

• src/mcp_prefect/task_run.py:30-129 (handler)

  The get_task_runs tool handler decorated with @mcp.tool, implementing the logic to fetch and filter Prefect task runs using the Prefect client, adding UI links, and returning as TextContent.

  @mcp.tool
  async def get_task_runs(
      limit: Optional[int] = None,
      offset: Optional[int] = None,
      task_name: Optional[str] = None,
      state_type: Optional[str] = None,
      state_name: 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 task runs with optional filtering.
      
      Args:
          limit: Maximum number of task runs to return
          offset: Number of task runs to skip
          task_name: Filter by task name
          state_type: Filter by state type (e.g., "RUNNING", "COMPLETED", "FAILED")
          state_name: Filter by state name
          tags: Filter by tags
          start_time_before: ISO formatted datetime string
          start_time_after: ISO formatted datetime string
          
      Returns:
          A list of task runs with their details
      """
      async with get_client() as client:
          # Build filter objects
          task_run_filter = None
          filter_components = []
          
          if task_name:
              filter_components.append(
                  TaskRunFilterName(like_=f"%{task_name}%")
              )
          
          if state_type:
              filter_components.append(
                  TaskRunFilterState(
                      type=TaskRunFilterStateType(any_=[state_type.upper()])
                  )
              )
          
          if state_name:
              filter_components.append(
                  TaskRunFilterState(
                      name=TaskRunFilterStateName(any_=[state_name])
                  )
              )
          
          if tags:
              filter_components.append(
                  TaskRunFilterTags(all_=tags)
              )
          
          if start_time_after or start_time_before:
              start_time_filter_args = {}
              if start_time_after:
                  start_time_filter_args["after_"] = datetime.fromisoformat(start_time_after)
              if start_time_before:
                  start_time_filter_args["before_"] = datetime.fromisoformat(start_time_before)
              filter_components.append(
                  TaskRunFilterStartTime(**start_time_filter_args)
              )
          
          # Combine filters if any exist
          if filter_components:
              # Create TaskRunFilter with the components
              # Note: You may need to adjust this based on how TaskRunFilter combines filters
              task_run_filter = TaskRunFilter()
              for component in filter_components:
                  if isinstance(component, TaskRunFilterName):
                      task_run_filter.name = component
                  elif isinstance(component, TaskRunFilterState):
                      task_run_filter.state = component
                  elif isinstance(component, TaskRunFilterTags):
                      task_run_filter.tags = component
                  elif isinstance(component, TaskRunFilterStartTime):
                      task_run_filter.start_time = component
          
          task_runs = await client.read_task_runs(
              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))]

• src/mcp_prefect/task_run.py:25-27 (helper)

  Helper function used by get_task_runs to generate UI URLs for task runs.

  def get_task_run_url(task_run_id: str) -> str:
      base_url = PREFECT_API_URL.replace("/api", "")
      return f"{base_url}/task-runs/{task_run_id}"