list_tasks
Retrieve all tasks within a specified project and domain to manage and track progress effectively.
Instructions
List all tasks in a project and domain.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| domain | Yes | ||
| project | Yes |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
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