Union MCP

Instructions

- Based on the prompt and inputs dictionary, determine the workflow to run
- Format the inputs dictionary so that it matches the workflow function signature
- Invoke the workflow

Args:
    project: Project to run the workflow in.
    domain: Domain to run the workflow in.
    name: Name of the task to run.
    inputs: A dictionary of inputs to the workflow.

Returns:
    A dictionary of outputs from the workflow.

Implementation Reference

• union_mcp/v1/server.py:64-95 (handler)

  The handler function decorated with @mcp.tool(), implementing the run_workflow tool. It executes a workflow in Union using the provided inputs, waits for completion, and returns the outputs and console URL.

  @mcp.tool()
  @require_auth
  def run_workflow(
      name: str,
      inputs: dict,
      project: str,
      domain: str,
      ctx: Context,
  ) -> tuple[dict, str]:
      """Run a workflow with natural language.
  
      - Based on the prompt and inputs dictionary, determine the workflow to run
      - Format the inputs dictionary so that it matches the workflow function signature
      - Invoke the workflow
  
      Args:
          project: Project to run the workflow in.
          domain: Domain to run the workflow in.
          name: Name of the task to run.
          inputs: A dictionary of inputs to the workflow.
  
      Returns:
          A dictionary of outputs from the workflow.
      """
      print(f"Running workflow {name} in project {project} and domain {domain}")
      remote = _remote(project, domain)
      workflow = remote.fetch_workflow(project=project, domain=domain, name=name)
      execution = remote.execute(workflow, inputs, project=project, domain=domain)
      execution = remote.wait(execution, poll_interval=timedelta(seconds=2))
      outputs = {k: v for k, v in execution.outputs.items() if v is not None}
      url = remote.generate_console_url(execution)
      return outputs, url