Workflow Management API

This document outlines the API endpoints for managing workflows in PySpur.

Create Workflow

Description: Creates a new workflow. If no definition is provided, creates a default workflow with an input node. For chatbots, creates a workflow with required input/output fields for handling chat interactions. The workflow name will be made unique if a workflow with the same name already exists. URL: /wf/ Method: POST Request Payload:

class WorkflowCreateRequestSchema(BaseModel):
    name: str  # Name of the workflow
    description: str = ""  # Description of the workflow
    definition: Optional[WorkflowDefinitionSchema] = None  # Definition of the workflow

Where WorkflowDefinitionSchema contains:

class WorkflowDefinitionSchema(BaseModel):
    nodes: List[WorkflowNodeSchema]  # List of nodes in the workflow
    links: List[WorkflowLinkSchema]  # List of links between nodes
    test_inputs: List[Dict[str, Any]] = []  # Test inputs for the workflow
    spur_type: SpurType = SpurType.WORKFLOW  # Type of workflow (WORKFLOW, CHATBOT, AGENT)

Response Schema:

class WorkflowResponseSchema(BaseModel):
    id: str  # Workflow ID
    name: str  # Name of the workflow
    description: Optional[str]  # Description of the workflow
    definition: WorkflowDefinitionSchema  # Definition of the workflow
    created_at: datetime  # When the workflow was created
    updated_at: datetime  # When the workflow was last updated

Update Workflow

Description: Updates an existing workflow’s definition, name, and description. The workflow definition is required for updates. This endpoint allows for modifying the structure and behavior of a workflow. URL: /wf/{workflow_id}/ Method: PUT Parameters:

workflow_id: str  # ID of the workflow to update

Request Payload: Same as Create Workflow Response Schema: Same as Create Workflow

List Workflows

Description: Lists all workflows with pagination support, ordered by creation date descending. Only valid workflows that can be properly validated are included in the response. URL: /wf/ Method: GET Query Parameters:

page: int  # Page number (default: 1, min: 1)
page_size: int  # Number of items per page (default: 10, min: 1, max: 100)

Response Schema:

List[WorkflowResponseSchema]

Get Workflow

Description: Retrieves a specific workflow by its ID, including its complete definition, metadata, and timestamps. URL: /wf/{workflow_id}/ Method: GET Parameters:

workflow_id: str  # ID of the workflow to retrieve

Response Schema: Same as Create Workflow

Reset Workflow

Description: Resets a workflow to its initial state with just an input node. This is useful when you want to start over with a workflow design without deleting and recreating it. URL: /wf/{workflow_id}/reset/ Method: PUT Parameters:

workflow_id: str  # ID of the workflow to reset

Response Schema: Same as Create Workflow

Delete Workflow

Description: Deletes a workflow and its associated test files. This operation is permanent and will remove all data related to the workflow, including test files stored in the file system. URL: /wf/{workflow_id}/ Method: DELETE Parameters:

workflow_id: str  # ID of the workflow to delete

Response: 204 No Content

Duplicate Workflow

Description: Creates a copy of an existing workflow with “(Copy)” appended to its name. This is useful for creating variations of a workflow without modifying the original. URL: /wf/{workflow_id}/duplicate/ Method: POST Parameters:

workflow_id: str  # ID of the workflow to duplicate

Response Schema: Same as Create Workflow

Get Workflow Output Variables

Description: Retrieves the output variables (leaf nodes) of a workflow, including their node IDs and variable names. This is useful for understanding what outputs are available from a workflow. URL: /wf/{workflow_id}/output_variables/ Method: GET Parameters:

workflow_id: str  # ID of the workflow

Response Schema:

List[Dict[str, str]]  # List of output variables with node IDs and variable names

Each dictionary in the list contains:

{
    "node_id": str,  # ID of the node
    "variable_name": str,  # Name of the output variable
    "prefixed_variable": str  # Variable name prefixed with node ID (node_id-variable_name)
}

Upload Test Files

Description: Uploads test files for a specific node in a workflow and returns their paths. The files are stored in a workflow-specific directory and can be used as inputs for testing the workflow. URL: /wf/upload_test_files/ Method: POST Form Data:

workflow_id: str  # ID of the workflow
files: List[UploadFile]  # List of files to upload
node_id: str  # ID of the node to associate files with

Response Schema:

Dict[str, List[str]]  # Dictionary mapping node ID to list of file paths

Example:

{
    "node_id": ["test_files/workflow_id/timestamp_filename.ext", ...]
}

Get Workflow Versions

Description: Retrieves all versions of a workflow, ordered by version number descending, with pagination support. This allows tracking the evolution of a workflow over time and reverting to previous versions if needed. URL: /wf/{workflow_id}/versions/ Method: GET Parameters:

workflow_id: str  # ID of the workflow
page: int  # Page number (default: 1, min: 1)
page_size: int  # Number of items per page (default: 10, min: 1, max: 100)

Response Schema:

List[WorkflowVersionResponseSchema]

Where WorkflowVersionResponseSchema contains:

class WorkflowVersionResponseSchema(BaseModel):
    version: int  # Version number
    name: str  # Name of the workflow version
    description: Optional[str]  # Description of the workflow version
    definition: Any  # Definition of the workflow version
    definition_hash: str  # Hash of the definition for tracking changes
    created_at: datetime  # When the version was created
    updated_at: datetime  # When the version was last updated

List Paused Workflows

Description: Lists all workflows that are currently in a paused state, with pagination support. This endpoint is useful for monitoring workflows that require human intervention. URL: /wf/paused_workflows/ Method: GET Query Parameters:

page: int  # Page number (default: 1, min: 1)
page_size: int  # Number of items per page (default: 10, min: 1, max: 100)

Response Schema:

List[PausedWorkflowResponseSchema]

Where PausedWorkflowResponseSchema contains:

class PausedWorkflowResponseSchema(BaseModel):
    run: RunResponseSchema  # Information about the workflow run
    current_pause: PauseHistoryResponseSchema  # Details about the current pause state
    workflow: WorkflowDefinitionSchema  # The workflow definition

Get Pause History

Description: Retrieves the pause history for a specific workflow run, showing when and why the workflow was paused, and any actions taken to resume it. URL: /wf/pause_history/{run_id}/ Method: GET Parameters:

run_id: str  # ID of the workflow run

Response Schema:

List[PauseHistoryResponseSchema]

Where PauseHistoryResponseSchema contains:

class PauseHistoryResponseSchema(BaseModel):
    id: str  # Synthetic ID for API compatibility
    run_id: str  # ID of the run
    node_id: str  # ID of the node where the pause occurred
    pause_message: Optional[str]  # Message explaining the pause reason
    pause_time: datetime  # When the workflow was paused
    resume_time: Optional[datetime]  # When the workflow was resumed (if applicable)
    resume_user_id: Optional[str]  # ID of the user who resumed the workflow
    resume_action: Optional[PauseAction]  # Action taken (APPROVE/DECLINE/OVERRIDE)
    input_data: Optional[Dict[str, Any]]  # Input data at the time of pause
    comments: Optional[str]  # Additional comments about the pause/resume

Process Pause Action

Description: Processes an action on a paused workflow, allowing for approval, decline, or override of a workflow that has been paused for human intervention. The workflow will resume execution based on the action taken. URL: /wf/process_pause_action/{run_id}/ Method: POST Parameters:

run_id: str  # ID of the workflow run

Request Payload:

class ResumeRunRequestSchema(BaseModel):
    inputs: Dict[str, Any]  # Human-provided inputs for the paused node
    user_id: str  # ID of the user resuming the workflow
    action: PauseAction  # Action taken (APPROVE/DECLINE/OVERRIDE)
    comments: Optional[str] = None  # Optional comments about the decision

Response Schema:

class RunResponseSchema(BaseModel):
    id: str  # Run ID
    workflow_id: str  # ID of the workflow
    workflow_version_id: Optional[str]  # ID of the workflow version
    workflow_version: Optional[WorkflowVersionResponseSchema]  # Details of the workflow version
    status: RunStatus  # Current status of the run
    start_time: datetime  # When the run started
    end_time: Optional[datetime]  # When the run ended (if completed)
    initial_inputs: Optional[Dict[str, Dict[str, Any]]]  # Initial inputs to the workflow
    outputs: Optional[Dict[str, Dict[str, Any]]]  # Outputs from the workflow
    tasks: List[TaskResponseSchema]  # List of tasks in the run
    parent_run_id: Optional[str]  # ID of the parent run (if applicable)
    run_type: str  # Type of run (e.g., "interactive")
    output_file_id: Optional[str]  # ID of the output file
    input_dataset_id: Optional[str]  # ID of the input dataset
    message: Optional[str]  # Additional information about the run
    duration: Optional[float]  # Duration of the run in seconds
    percentage_complete: float  # Percentage of tasks completed

​Workflow Management API

​Create Workflow

​Update Workflow

​List Workflows

​Get Workflow

​Reset Workflow

​Delete Workflow

​Duplicate Workflow

​Get Workflow Output Variables

​Upload Test Files

​Get Workflow Versions

​List Paused Workflows

​Get Pause History

​Process Pause Action

Workflow Management API

Create Workflow

Update Workflow

List Workflows

Get Workflow

Reset Workflow

Delete Workflow

Duplicate Workflow

Get Workflow Output Variables

Upload Test Files

Get Workflow Versions

List Paused Workflows

Get Pause History

Process Pause Action