td_list_workflows
Monitor Treasure Data workflows to identify failed jobs, track execution status, and audit data pipeline health across all projects.
Instructions
List all workflows to monitor executions and find failed jobs.
Shows workflows across all projects with their latest execution status.
Essential for monitoring data pipeline health and finding issues.
Common scenarios:
- Check which workflows are failing (status_filter="error")
- Monitor currently running workflows (status_filter="running")
- Find workflows by name (use search parameter)
- Get overview of all scheduled jobs
- Audit workflow execution patterns
Filter options: status ('success', 'error', 'running'), search by name.
Set verbose=True for execution history. Limit count to avoid token issues.Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| verbose | No | ||
| count | No | ||
| include_system | No | ||
| status_filter | No | ||
| search | No |
Implementation Reference
- td_mcp_server/mcp_impl.py:591-703 (handler)The primary handler for the 'td_list_workflows' tool. It creates a TreasureDataClient, fetches workflows using client.get_workflows, applies filters for system workflows, status, and search terms, and returns formatted results in verbose or summary mode. Decorated with @mcp.tool() for automatic registration.
@mcp.tool() async def td_list_workflows( verbose: bool = False, count: int = 50, include_system: bool = False, status_filter: str | None = None, search: str | None = None, ) -> dict[str, Any]: """List all workflows to monitor executions and find failed jobs. Shows workflows across all projects with their latest execution status. Essential for monitoring data pipeline health and finding issues. Common scenarios: - Check which workflows are failing (status_filter="error") - Monitor currently running workflows (status_filter="running") - Find workflows by name (use search parameter) - Get overview of all scheduled jobs - Audit workflow execution patterns Filter options: status ('success', 'error', 'running'), search by name. Set verbose=True for execution history. Limit count to avoid token issues. """ client = _create_client(include_workflow=True) if isinstance(client, dict): return client try: workflows = client.get_workflows(count=min(count, 12000), all_results=True) # Filter out system workflows if requested if not include_system: workflows = [ w for w in workflows if not any( meta.key == "sys" for meta in w.project.model_dump().get("metadata", []) ) ] # Filter by status if requested if status_filter: filtered_workflows = [] for workflow in workflows: if workflow.latest_sessions: last_status = workflow.latest_sessions[0].last_attempt.status if last_status == status_filter: filtered_workflows.append(workflow) workflows = filtered_workflows # Filter by search term if requested if search: search_lower = search.lower() filtered_workflows = [] for workflow in workflows: workflow_name = workflow.name.lower() project_name = workflow.project.name.lower() if search_lower in workflow_name or search_lower in project_name: filtered_workflows.append(workflow) workflows = filtered_workflows if verbose: # Return full workflow details including sessions return { "workflows": [ { "id": w.id, "name": w.name, "project": { "id": w.project.id, "name": w.project.name, }, "timezone": w.timezone, "schedule": w.schedule, "latest_sessions": [ { "session_time": s.session_time, "status": s.last_attempt.status, "success": s.last_attempt.success, "duration": None, # Would need date parsing } for s in w.latest_sessions[:3] # Show last 3 sessions ], } for w in workflows ] } else: # Return summary information return { "workflows": [ { "id": w.id, "name": w.name, "project": w.project.name, "last_status": ( w.latest_sessions[0].last_attempt.status if w.latest_sessions else "no_runs" ), "scheduled": w.schedule is not None, } for w in workflows ], "total_count": len(workflows), } except (ValueError, requests.RequestException) as e: return _format_error_response(f"Failed to retrieve workflows: {str(e)}") except Exception as e: return _format_error_response( f"Unexpected error while retrieving workflows: {str(e)}" ) - td_mcp_server/api.py:517-604 (helper)Supporting method in TreasureDataClient class that performs the actual API calls to retrieve workflows, supporting pagination and all_results mode by querying the /console/workflows endpoint.
def get_workflows( self, count: int = 100, all_results: bool = False, page: int = 1, ) -> list[Workflow]: """ Retrieve a list of workflows across all projects. This method retrieves workflows from the Treasure Data workflow console API. Workflows are the actual executable units that contain tasks defined in Digdag files (.dig). Each workflow belongs to a project and can have multiple sessions (execution instances). Args: count: Maximum number of workflows to retrieve per page (defaults to 100) all_results: If True, retrieves all workflows across multiple pages page: Page number for pagination (defaults to 1) Returns: A list of Workflow objects Raises: requests.HTTPError: If the API returns an error response """ if all_results: # Retrieve all workflows by iterating through pages all_workflows = [] current_page = 1 per_page = min(count, 1000) # Use reasonable page size while True: params = { "count": per_page, "page": current_page, "order": "asc", "sessions": 5, # Include last 5 sessions for each workflow "output": "simple", "project_type": "user", } response = requests.get( f"{self.workflow_base_url}/console/workflows", headers=self.headers, params=params, ) response.raise_for_status() data = response.json() workflows = [ Workflow(**workflow) for workflow in data.get("workflows", []) ] if not workflows: # No more workflows on this page break all_workflows.extend(workflows) # Check if we've reached the desired count if len(all_workflows) >= count: return all_workflows[:count] current_page += 1 return all_workflows else: # Single page request params = { "count": count, "page": page, "order": "asc", "sessions": 5, # Include last 5 sessions for each workflow "output": "simple", "project_type": "user", } response = requests.get( f"{self.workflow_base_url}/console/workflows", headers=self.headers, params=params, ) response.raise_for_status() data = response.json() workflows = [Workflow(**workflow) for workflow in data.get("workflows", [])] return workflows