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)