create_flow_run
Start a workflow execution by creating a new flow run for a specified deployment, with optional parameters to customize the run.
Instructions
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).Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| deployment_id | Yes | ||
| parameters | No |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- prefect_mcp_server_pkg/server.py:438-448 (handler)Primary handler for the "create_flow_run" MCP tool (legacy). Delegates to the core implementation function.
@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) - prefect_mcp_server_pkg/server.py:351-384 (handler)Core implementation of flow run creation, registered as a separate tool "create_flow_run_from_deployment". Called by the legacy "create_flow_run" tool.
@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)}"}