Union MCP

Instructions

Implementation Reference

• union_mcp/v2/server.py:68-76 (handler)

  MCP tool handler and registration for 'get_task'. Initializes Flyte and delegates to resources.get_task to fetch and return task details as dict.

  @mcp.tool()
  @require_auth
  async def get_task(name: str, project: str, domain: str, ctx: Context) -> dict:
      """Get a union task."""
      print(f"Getting task {name} in project {project} and domain {domain}")
      _init(project, domain)
      task = await resources.get_task(name, project, domain)
      return task.to_dict()

• union_mcp/v2/resources.py:22-35 (helper)

  Core helper function that retrieves a Flyte Task using flyte.remote.Task.get and fetches its details.

  async def get_task(
      name: str,
      project: str | None = None,
      domain: str | None = None,
      version: str | None = None,
  ) -> flyte.remote.Task:
      return flyte.remote.Task.get(
          name=name,
          project=project,
          domain=domain,
          version=version,
          auto_version="latest" if version is None else None,
      ).fetch()

• union_mcp/v1/server.py:98-106 (handler)

  MCP tool handler and registration for 'get_task' in v1. Uses UnionRemote to fetch task and returns string representation.

  @mcp.tool()
  @require_auth
  def get_task(name: str, project: str, domain: str, ctx: Context) -> str:
      """Get a union task."""
      print(f"Getting task {name} in project {project} and domain {domain}")
      remote = _remote(project, domain)
      task = remote.fetch_task(name=name, project=project, domain=domain)
      return str(task)