get_deployments
Retrieve and filter deployment lists from Prefect workflows using criteria like flow name, tags, schedule status, or work queue to manage and monitor automated processes.
Instructions
Get a list of deployments with optional filtering.
Args: limit: Maximum number of deployments to return offset: Number of deployments to skip flow_name: Filter by flow name name: Filter by deployment name tags: Filter by tags is_schedule_active: Filter by schedule active status work_queue_name: Filter by work queue name
Returns: A list of deployments with their details
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| flow_name | No | ||
| is_schedule_active | No | ||
| limit | No | ||
| name | No | ||
| offset | No | ||
| tags | No | ||
| work_queue_name | No |
Implementation Reference
- src/mcp_prefect/deployment.py:16-90 (handler)The @mcp.tool decorated handler function that implements the get_deployments tool logic. It queries Prefect deployments using filters and returns results with UI links.@mcp.tool async def get_deployments( limit: Optional[int] = None, offset: Optional[int] = None, flow_name: Optional[str] = None, name: Optional[str] = None, tags: Optional[List[str]] = None, is_schedule_active: Optional[bool] = None, work_queue_name: Optional[str] = None, ) -> List[Union[types.TextContent, types.ImageContent, types.EmbeddedResource]]: """ Get a list of deployments with optional filtering. Args: limit: Maximum number of deployments to return offset: Number of deployments to skip flow_name: Filter by flow name name: Filter by deployment name tags: Filter by tags is_schedule_active: Filter by schedule active status work_queue_name: Filter by work queue name Returns: A list of deployments with their details """ try: async with get_client() as client: # Build deployment filter deployment_filter = None if any([name, tags, is_schedule_active, work_queue_name]): from prefect.client.schemas.filters import DeploymentFilter filter_dict = {} if name: filter_dict["name"] = {"like_": f"%{name}%"} if tags: filter_dict["tags"] = {"all_": tags} if is_schedule_active is not None: filter_dict["is_schedule_active"] = {"eq_": is_schedule_active} if work_queue_name: filter_dict["work_queue_name"] = {"eq_": work_queue_name} deployment_filter = DeploymentFilter(**filter_dict) # Build flow filter if flow_name is specified flow_filter = None if flow_name: from prefect.client.schemas.filters import FlowFilter flow_filter = FlowFilter(name={"like_": f"%{flow_name}%"}) # Query using proper filter objects deployments = await client.read_deployments( deployment_filter=deployment_filter, flow_filter=flow_filter, limit=limit, offset=offset, ) # Add UI links to each deployment deployments_result = { "deployments": [ { **deployment.model_dump(), "ui_url": get_deployment_url(str(deployment.id)) } for deployment in deployments ] } return [types.TextContent(type="text", text=str(deployments_result))] except Exception as e: # Add proper error handling return [types.TextContent(type="text", text=str({"error": str(e)}))]
- src/mcp_prefect/main.py:46-48 (registration)Conditional import of the deployment module in main.py, which triggers registration of all deployment tools including get_deployments via their @mcp.tool decorators.if APIType.DEPLOYMENT.value in apis: info("Loading Deployment API...") from . import deployment
- src/mcp_prefect/deployment.py:11-13 (helper)Helper function used within get_deployments (and other deployment tools) to generate UI URLs for deployments.def get_deployment_url(deployment_id: str) -> str: base_url = PREFECT_API_URL.replace("/api", "") return f"{base_url}/deployments/{deployment_id}"