list_deployments
Retrieve and filter deployment records from Prefect workflows to monitor and manage automation processes.
Instructions
Get a list of deployments from the Prefect API.
Args:
limit: Maximum number of deployments to return (default 20).
offset: Number of deployments to skip (default 0).
flow_id: Optional ID of the flow to filter deployments by.Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| limit | No | ||
| offset | No | ||
| flow_id | No |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- prefect_mcp_server_pkg/server.py:298-322 (handler)The async handler function for the 'list_deployments' MCP tool. It retrieves a list of Prefect deployments using the Prefect client, supports pagination with limit/offset, and optional filtering by flow_id. The @mcp.tool() decorator registers this function as an MCP tool.
@mcp.tool() async def list_deployments( ctx: Context, limit: int = 20, offset: int = 0, flow_id: Optional[str] = None ) -> Dict[str, Any]: """Get a list of deployments from the Prefect API. Args: limit: Maximum number of deployments to return (default 20). offset: Number of deployments to skip (default 0). flow_id: Optional ID of the flow to filter deployments by. """ filter_dict = {} if flow_id: filter_dict["flow_id"] = {"equals": flow_id} async with get_client() as client: deployment_filter = DeploymentFilter(**filter_dict) if filter_dict else None deployments = await client.read_deployments( deployment_filter=deployment_filter, limit=limit, offset=offset ) return { "deployments": [depl.model_dump() for depl in deployments], "count": len(deployments), }