list_workflows
Retrieve all workflows available in a specified project and domain to manage and utilize Union automation processes.
Instructions
List all workflows 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/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 ]