Union MCP

Instructions

Implementation Reference

• union_mcp/v2/server.py:100-110 (handler)

  MCP tool handler for list_tasks in Union v2 server, which initializes Flyte context and delegates to resources.list_tasks.

  @mcp.tool()
  @require_auth
  async def list_tasks(
      project: str,
      domain: str,
      ctx: Context,
  ) -> list[dict]:
      """List all tasks in a project and domain."""
      _init(project, domain)
      print(f"Listing tasks in project {project} and domain {domain}")
      return [task.to_dict() for task in await resources.list_tasks(project, domain)]

• union_mcp/v1/server.py:132-142 (handler)

  MCP tool handler for list_tasks in Union v1 server, which creates remote client and delegates to resources.list_tasks.

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

• union_mcp/v2/resources.py:47-54 (helper)

  Helper function in v2 resources that lists tasks using Flyte's Task.listall and fetches details.

  async def list_tasks(
      project: str | None = None,
      domain: str | None = None,
  ) -> list[flyte.remote.Task]:
      tasks = []
      for task in flyte.remote.Task.listall(project=project, domain=domain):
          tasks.append(await get_task(task.name, project=project, domain=domain))
      return tasks

• union_mcp/v1/resources.py:29-43 (helper)

  Helper function in v1 resources that lists tasks using UnionRemote client and converts to TaskMetadata.

  def list_tasks(remote: union.UnionRemote, project: str, domain: str) -> list[TaskMetadata]:
      from flytekit.models.common import NamedEntityIdentifier
  
      id = NamedEntityIdentifier(project=project, domain=domain)
      task_models, _ = remote.client.list_tasks_paginated(id, limit=100)
      tasks = [t.to_flyte_idl() for t in task_models]
      return [
          TaskMetadata(
              name=task.id.name,
              description=task.short_description,
              inputs=proto_to_json(task.closure.compiled_task.template.interface.inputs),
              outputs=proto_to_json(task.closure.compiled_task.template.interface.outputs),
          )
          for task in tasks
      ]

• union_mcp/v1/resources.py:11-16 (schema)

  Pydantic schema for TaskMetadata used in v1 list_tasks responses.

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