Treasure Data MCP Server

Instructions

Implementation Reference

• td_mcp_server/mcp_impl.py:591-704 (handler)

  The primary handler function for the 'td_list_workflows' tool. It uses TreasureDataClient to fetch workflows, applies filters for system workflows, status, and search terms, and returns formatted results based on verbosity.
  @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)}" )