start_workflow_execution

Initiate workflow execution by ID with input data to create a queued job and obtain a job ID for tracking.

Instructions

Start a workflow execution by its ID and return the job ID. This will create a new job and add it to the execution queue. This call will return a job ID that can be used to get the job details later. The input data is a list of name-value pairs, each containing a name and value.

Input Schema

TableJSON Schema

Name	Required	Description	Default
`workflow_id`	Yes
`input_data`	No

Implementation Reference

src/tools.py:214-242 (handler)

Core handler function in AYXMCPTools class that validates input, enqueues the workflow job using Alteryx API, and returns the job response.

def start_workflow_execution(self, workflow_id: str, input_data: list[InputData] = None):
    """Start a workflow execution by its ID and return the job ID. This will create a new job and add it to the execution queue.
    This call will return a job ID that can be used to get the job details. Once the job is executed, 
    the results can be retrieved via the produced JobID
    The input data is a list of name-value pairs, each containing a name and value."""
    try:
        workflow = self.workflows_api.workflows_get_workflow(workflow_id)
        if not workflow:
            return "Error: Workflow not found"
        questions = self.workflows_api.workflows_get_workflow_questions(workflow_id)
        if (not questions or len(questions) == 0) and (input_data):
            return "Error: Workflow has no questions, input data not allowed"
        if questions and len(questions) > 0:
            for question in questions:
                if question.name not in [item.name for item in input_data]:
                    return f"Error: Input data must contain the question '{question.name}'"
                
        # Convert InputData objects to AppValue objects
        app_values = None
        if input_data:
            app_values = [server_client.AppValue(name=item.name, value=item.value) for item in input_data]
                
        # Proper type conversion
        workflow = server_client.WorkflowView(workflow)
        contract = server_client.EnqueueJobContract(worker_tag=workflow.worker_tag, questions=app_values)
        api_response = self.workflows_api.workflows_enqueue(workflow_id, contract)
        return pprint.pformat(api_response)
    except ApiException as e:
        return f"Error: {e}"

src/tools.py:15-18 (schema)
Pydantic BaseModel defining the structure for input data parameters (list of name-value pairs) used in workflow execution.
```
class InputData(BaseModel):
    name: str
    value: str
```

src/mcp_server.py:207-214 (registration)

MCP tool registration decorator (@app.tool()) that wraps and delegates to the core handler in tools.py.

@self.app.tool()
def start_workflow_execution(workflow_id: str, input_data: list[InputData] = None):
    """Start a workflow execution by its ID and return the job ID. 
    This will create a new job and add it to the execution queue.
    This call will return a job ID that can be used to get the job details later. 
    The input data is a list of name-value pairs, each containing a name and value."""
    return self.tools.start_workflow_execution(workflow_id, input_data)

AYX-MCP-Wrapper