AYX-MCP-Wrapper

Instructions

Implementation Reference

• src/tools.py:518-587 (handler)

  The main handler function that implements the logic for get_job_output_data. It checks if the job exists and is completed, then downloads each output file from the Alteryx API, saves them to the temp directory with appropriate extensions based on available formats, and returns the list of saved file paths.
  def get_job_output_data(self, job_id: str): """Get the output data for a job""" try: # check if job exists job = self.jobs_api.jobs_get_job_v3(job_id) if not job: return "Error: Job not found" # check if job is completed if job.status != "Completed": return "Error: Job is not completed" temp_directory = self.configuration.temp_directory # normalize the temp directory temp_directory = os.path.normpath(temp_directory) if not os.path.exists(temp_directory): os.makedirs(temp_directory) all_output_files = [] for output in job.outputs: output_id = output.id file_name = output.file_name available_output_types = output.available_formats # get file name with extension from file_name # Extract base name without extension if it exists base_name = os.path.splitext(os.path.basename(file_name))[0] # Get the file extension from the file name raw_file_extension = os.path.splitext(os.path.basename(file_name))[1] # Map output format to file extension format_extension_map = { 'Raw': raw_file_extension if raw_file_extension else '.txt', 'Yxdb': '.yxdb', 'Shp': '.shp', 'Kml': '.kml', 'Tab': '.tab', 'Mif': '.mif', 'Dbf': '.dbf', 'Csv': '.csv', 'Pdf': '.pdf', 'Docx': '.docx', 'Xlsx': '.xlsx', 'Html': '.html', 'Tde': '.tde', 'Zip': '.zip' } # Get the extension for the first available format output_format = available_output_types[0] if available_output_types else 'Raw' file_extension = format_extension_map.get(output_format, raw_file_extension) file_name_with_extension = base_name + file_extension # get the output data api_response = self.jobs_api.jobs_get_output_file(job_id, output_id, output_format) # Convert to bytes if it's a string if isinstance(api_response, str): api_response_bytes = api_response.encode('utf-8') else: api_response_bytes = api_response with open(f"{temp_directory}/{job_id}_{output_id}_{file_name_with_extension}", "wb") as f: f.write(api_response_bytes) all_output_files.append(f"{temp_directory}/{job_id}_{output_id}_{file_name_with_extension}") return f"Output files saved to: {pprint.pformat(all_output_files)} \n\n" except ApiException as e: return f"Error: {e}"
• src/mcp_server.py:297-302 (registration)

  The MCP tool registration using @self.app.tool() decorator. This wrapper function delegates the call to the underlying tools instance's get_job_output_data method.
  @self.app.tool() def get_job_output_data(job_id: str): """Get the output data generated by a job. This will return a list of file paths to the output data. The output data is stored in the temp directory of the server.""" return self.tools.get_job_output_data(job_id)