search_flows
Find Prefect workflows by name or tags to manage automation processes efficiently.
Instructions
Search for flows by name and/or tags.
Args:
name: Optional name to search for (case-insensitive contains match).
tags: Optional list of tags to filter by.
limit: Maximum number of flows to return (default 20).Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| name | No | ||
| tags | No | ||
| limit | No |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- prefect_mcp_server_pkg/server.py:112-138 (handler)Implements the 'search_flows' MCP tool. Decorated with @mcp.tool() for automatic registration and schema generation from type hints. Searches Prefect flows using FlowFilter for name (contains) and tags (all), retrieves via client.read_flows, and returns list with count.
@mcp.tool() async def search_flows( ctx: Context, name: Optional[str] = None, tags: Optional[List[str]] = None, limit: int = 20, ) -> Dict[str, Any]: """Search for flows by name and/or tags. Args: name: Optional name to search for (case-insensitive contains match). tags: Optional list of tags to filter by. limit: Maximum number of flows to return (default 20). """ filter_dict = {} if name: filter_dict["name"] = {"contains": name} if tags: filter_dict["tags"] = {"all_": tags} async with get_client() as client: flow_filter = FlowFilter(**filter_dict) flows = await client.read_flows(flow_filter=flow_filter, limit=limit) return {"flows": [flow.model_dump() for flow in flows], "count": len(flows)}