start_workflow_execution
Initiate a workflow execution by its ID using input data as name-value pairs, returning a job ID for tracking and managing the process.
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 |
|---|---|---|---|
| input_data | No | ||
| workflow_id | Yes |
Implementation Reference
- src/tools.py:214-243 (handler)Core implementation of the start_workflow_execution tool. Validates workflow existence and input data against questions, converts inputs to AppValue format, enqueues the job using Alteryx V3 API, and returns the formatted 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/mcp_server.py:207-214 (registration)MCP tool registration decorator and wrapper function that delegates to the AYXMCPTools instance's start_workflow_execution method.@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)
- src/tools.py:15-18 (schema)Pydantic model defining the schema for input data parameters (name-value pairs) used in workflow execution.class InputData(BaseModel): name: str value: str