Prefect MCP Server

Instructions

Implementation Reference

• src/mcp_prefect/flow.py:17-88 (handler)

  The handler function for the 'get_flows' MCP tool. It queries Prefect flows using the client with optional filters and returns results with UI links.

  @mcp.tool
  async def get_flows(
      limit: Optional[int] = None,
      offset: Optional[int] = None,
      flow_name: Optional[str] = None,
      tags: Optional[List[str]] = None,
      created_after: Optional[str] = None,
      created_before: Optional[str] = None,
  ) -> List[Union[types.TextContent, types.ImageContent, types.EmbeddedResource]]:
      """
      Get a list of flows with optional filtering.
      
      Args:
          limit: Maximum number of flows to return
          offset: Number of flows to skip
          flow_name: Filter flows by name
          tags: Filter flows by tags
          created_after: ISO formatted datetime string for filtering flows created after this time
          created_before: ISO formatted datetime string for filtering flows created before this time
          
      Returns:
          A list of flows with their details
      """
      try:
          async with get_client() as client:
              # Build flow filter
              flow_filter = None
              if any([flow_name, tags, created_after, created_before]):
                  from prefect.client.schemas.filters import FlowFilter
                  
                  filter_dict = {}
                  if flow_name:
                      filter_dict["name"] = {"like_": f"%{flow_name}%"}
                  if tags:
                      filter_dict["tags"] = {"all_": tags}
                  
                  # Handle date filters
                  if created_after or created_before:
                      created_filters = {}
                      if created_after:
                          created_filters["ge_"] = created_after
                      if created_before:
                          created_filters["le_"] = created_before
                      filter_dict["created"] = created_filters
                  
                  flow_filter = FlowFilter(**filter_dict)
              
              # Query using proper filter object
              flows = await client.read_flows(
                  flow_filter=flow_filter,
                  limit=limit,
                  offset=offset,
              )
              
              # Handle empty results
              if not flows:
                  return [types.TextContent(type="text", text=str({"flows": []}))]
              
              # Add UI links to each flow
              flows_with_links = []
              for flow in flows:
                  flow_dict = flow.model_dump()
                  flow_dict["ui_url"] = get_flow_url(str(flow.id))
                  flows_with_links.append(flow_dict)
                  
              flows_result = {"flows": flows_with_links}
              
              return [types.TextContent(type="text", text=str(flows_result))]        
      except Exception as e:
          error_message = f"Error fetching flows: {str(e)}"
          return [types.TextContent(type="text", text=error_message)]

• src/mcp_prefect/main.py:37-40 (registration)

  The import statement in main.py that loads the flow module, thereby registering the @mcp.tool decorated get_flows function.

  # Import modules to register their decorated tools
  if APIType.FLOW.value in apis:
      info("Loading Flow API...")
      from . import flow

• src/mcp_prefect/flow.py:11-14 (helper)

  Helper function used within get_flows to generate UI URLs for flows.

  def get_flow_url(flow_id: str) -> str:
      """Generate a UI URL for a flow."""
      base_url = PREFECT_API_URL.replace("/api", "")
      return f"{base_url}/flows/{flow_id}"