Prefect MCP Server

Instructions

Args:
    deployment_id: ID of the deployment to create a run for.
    parameters: Dictionary with parameters for the flow run (optional).

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)}"}