Prefect MCP Server

Instructions

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).

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()

• prefect_mcp_server_pkg/server.py:438-449 (helper)

  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)