Union MCP

Instructions

Implementation Reference

• union_mcp/v1/server.py:145-156 (handler)

  MCP tool handler for 'list_workflows', registered via @mcp.tool(), takes project and domain, delegates to resources.list_workflows

  @mcp.tool()
  @require_auth
  def list_workflows(
      project: str,
      domain: str,
      ctx: Context,
  ) -> list[resources.WorkflowMetadata]:
      """List all workflows in a project and domain."""
      remote = _remote(project, domain)
      print(f"Listing workflows in project {project} and domain {domain}")
      return resources.list_workflows(remote, project, domain)

• union_mcp/v1/resources.py:17-22 (schema)

  Pydantic schema/model for WorkflowMetadata returned by list_workflows

  class WorkflowMetadata(BaseModel):
      name: str
      description: str
      inputs: dict
      outputs: dict

• union_mcp/v1/resources.py:46-60 (helper)

  Helper function implementing the core logic: fetches workflows via remote.client.list_workflows_paginated and maps to WorkflowMetadata instances

  def list_workflows(remote: union.UnionRemote, project: str, domain: str) -> list[WorkflowMetadata]:
      from flytekit.models.common import NamedEntityIdentifier
  
      id = NamedEntityIdentifier(project=project, domain=domain)
      workflow_models, _ = remote.client.list_workflows_paginated(id, limit=100)
      workflows = [w.to_flyte_idl() for w in workflow_models]
      return [
          WorkflowMetadata(
              name=workflow.id.name,
              description=workflow.short_description,
              inputs=proto_to_json(workflow.closure.compiled_workflow.primary.template.interface.inputs),
              outputs=proto_to_json(workflow.closure.compiled_workflow.primary.template.interface.outputs),
          )
          for workflow in workflows
      ]