Prefect MCP Server

Instructions

Args:
    name: Name of the flow to retrieve.

Implementation Reference

• prefect_mcp_server_pkg/server.py:74-97 (handler)

  The main handler function for the 'get_flow_by_name' MCP tool. It is decorated with @mcp.tool(), which also serves as its registration. Retrieves a Prefect flow by name using the Prefect API client, returning the first matching flow or an error.

  @mcp.tool()
  async def get_flow_by_name(ctx: Context, name: str) -> Dict[str, Any]:
      """Get a flow by its name.
  
      Args:
          name: Name of the flow to retrieve.
      """
      if not name:
          return {"error": "Missing required argument: name"}
  
      async with get_client() as client:
          try:
              # Use correct flow_filter parameter
              flow_filter = FlowFilter(name=FlowFilterName(any_=[name]))
              flows = await client.read_flows(flow_filter=flow_filter)
  
              if not flows:
                  return {"error": f"No flow found with name: {name}"}
  
              # Return the first matching flow
              return {"flow": flows[0].model_dump()}
          except Exception as e:
              return {"error": f"Failed to get flow: {str(e)}"}