Method execute_workflow

packages/python-sdk/simstudio/__init__.py:150–264

Execute a workflow with optional input data. If async_execution is True, returns immediately with a task ID. File objects in input will be automatically detected and converted to base64. Args: workflow_id: The ID of the workflow to execute i

(
        self,
        workflow_id: str,
        input: Optional[Any] = None,
        *,
        timeout: float = 30.0,
        stream: Optional[bool] = None,
        selected_outputs: Optional[list] = None,
        async_execution: Optional[bool] = None
    )

        return value

    def execute_workflow(
        self,
        workflow_id: str,
        input: Optional[Any] = None,
        *,
        timeout: float = 30.0,
        stream: Optional[bool] = None,
        selected_outputs: Optional[list] = None,
        async_execution: Optional[bool] = None
    ) -> Union[WorkflowExecutionResult, AsyncExecutionResult]:
        """
        Execute a workflow with optional input data.
        If async_execution is True, returns immediately with a task ID.

        File objects in input will be automatically detected and converted to base64.

        Args:
            workflow_id: The ID of the workflow to execute
            input: Input data to pass to the workflow. Can be a dict (spread at root level),
                   primitive value (string, number, bool), or list (wrapped in 'input' field).
                   File-like objects within dicts are automatically converted to base64.
            timeout: Timeout in seconds (default: 30.0)
            stream: Enable streaming responses (default: None)
            selected_outputs: Block outputs to stream (e.g., ["agent1.content"])
            async_execution: Execute asynchronously (default: None)

        Returns:
            WorkflowExecutionResult or AsyncExecutionResult object

        Raises:
            SimStudioError: If the workflow execution fails
        """
        url = f"{self.base_url}/api/workflows/{workflow_id}/execute"

        # Build headers - async execution uses X-Execution-Mode header
        headers = self._session.headers.copy()
        if async_execution:
            headers['X-Execution-Mode'] = 'async'

        try:
            # Build JSON body - spread dict inputs at root level, wrap primitives/lists in 'input' field
            body = {}
            if input is not None:
                if isinstance(input, dict):
                    # Dict input: spread at root level (matches curl/API behavior)
                    body = input.copy()
                else:
                    # Primitive or list input: wrap in 'input' field
                    body = {'input': input}

            # Convert any file objects in the input to base64 format
            body = self._convert_files_to_base64(body)

            if stream is not None:
                body['stream'] = stream
            if selected_outputs is not None:
                body['selectedOutputs'] = selected_outputs