create_flow_run_from_deployment
Initiate a flow run for a specific deployment in Prefect MCP Server by providing deployment ID, optional parameters, name, and timeout settings.
Instructions
Create a new flow run for the specified deployment.
Args:
deployment_id: ID of the deployment or name in format 'flow_name/deployment_name'.
parameters: Dictionary with parameters for the flow run (optional).
name: Optional name for the flow run.
timeout: Timeout in seconds, 0 means no waiting for completion (default 0).
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| deployment_id | Yes | ||
| name | No | ||
| parameters | No | ||
| timeout | No |
Implementation Reference
- prefect_mcp_server_pkg/server.py:351-384 (handler)The primary handler function for the 'create_flow_run_from_deployment' tool. It is decorated with @mcp.tool() for registration and implements the core logic using Prefect's run_deployment function.@mcp.tool() async def create_flow_run_from_deployment( ctx: Context, deployment_id: str, parameters: Optional[Dict[str, Any]] = None, name: Optional[str] = None, timeout: int = 0, ) -> Dict[str, Any]: """Create a new flow run for the specified deployment. Args: deployment_id: ID of the deployment or name in format 'flow_name/deployment_name'. parameters: Dictionary with parameters for the flow run (optional). name: Optional name for the flow run. timeout: Timeout in seconds, 0 means no waiting for completion (default 0). """ if not deployment_id: return {"error": "Missing required argument: deployment_id"} from prefect.deployments import run_deployment try: # Создаем flow run с помощью функции run_deployment result = await run_deployment( name=deployment_id, # В документации это "name", а не "deployment_id" parameters=parameters or {}, timeout=timeout, flow_run_name=name, ) return {"flow_run_id": str(result)} except Exception as e: return {"error": f"Failed to create flow run: {str(e)}"}
- prefect_mcp_server_pkg/server.py:351-351 (registration)The @mcp.tool() decorator registers the create_flow_run_from_deployment function as an MCP tool.@mcp.tool()
- Legacy wrapper tool 'create_flow_run' that calls the main 'create_flow_run_from_deployment' handler.@mcp.tool() async def create_flow_run( ctx: Context, deployment_id: str, parameters: Optional[Dict[str, Any]] = None ) -> Dict[str, Any]: """Create a new flow run for the specified deployment (Legacy). Args: deployment_id: ID of the deployment to create a run for. parameters: Dictionary with parameters for the flow run (optional). """ return await create_flow_run_from_deployment(ctx, deployment_id, parameters)