create_flow_run_from_deployment
Trigger a workflow execution by launching a flow run from a Prefect deployment, with optional parameters, custom naming, 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 | ||
| parameters | No | ||
| name | No | ||
| timeout | No |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- prefect_mcp_server_pkg/server.py:351-384 (handler)The @mcp.tool()-decorated async handler function implementing the core logic of creating a flow run from a deployment 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 delegates to 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)