get_flow
Retrieve detailed information about a specific workflow by providing its unique identifier, enabling users to inspect flow configurations and status.
Instructions
Get details of a specific flow by ID.
Args: flow_id: The flow UUID
Returns: Flow details
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| flow_id | Yes |
Implementation Reference
- src/mcp_prefect/flow.py:90-125 (handler)The handler function for the 'get_flow' tool. It retrieves details of a specific flow by its ID using the Prefect client, adds a UI URL, and returns the information as text content.
@mcp.tool async def get_flow( flow_id: str, ) -> List[Union[types.TextContent, types.ImageContent, types.EmbeddedResource]]: """ Get details of a specific flow by ID. Args: flow_id: The flow UUID Returns: Flow details """ try: async with get_client() as client: # Validate flow_id try: flow_uuid = UUID(flow_id) except ValueError: return [types.TextContent( type="text", text=f"Invalid flow ID format: {flow_id}. Must be a valid UUID." )] flow = await client.read_flow(flow_uuid) # Add UI link flow_dict = flow.model_dump() flow_dict["ui_url"] = get_flow_url(flow_id) return [types.TextContent(type="text", text=str(flow_dict))] except Exception as e: error_message = f"Error fetching flow {flow_id}: {str(e)}" return [types.TextContent(type="text", text=error_message)] - src/mcp_prefect/main.py:38-41 (registration)The import statement in main.py that loads the flow module, thereby registering the 'get_flow' tool via its @mcp.tool decorator.
if APIType.FLOW.value in apis: info("Loading Flow API...") from . import flow - src/mcp_prefect/flow.py:11-14 (helper)Helper function used by 'get_flow' to generate a UI URL for the flow.
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}"