​
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