td_list_sessions
Monitor recent workflow executions to check status and identify failures. Filter by workflow ID to view history or see all executions.
Instructions
List recent workflow executions to monitor status and find failures.
Shows recent scheduled runs (sessions) with their execution status. Filter by
workflow ID to see history of a specific workflow, or leave empty for all.
Common scenarios:
- Check which workflows ran recently and their status
- Find failed executions that need investigation
- Monitor execution patterns for a specific workflow
- Get session IDs for detailed analysis
Returns list with workflow names, execution times, and success/failure status.Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| workflow_id | No | ||
| count | No |
Implementation Reference
- td_mcp_server/execution_tools.py:83-135 (handler)The main asynchronous handler function for the 'td_list_sessions' tool. It lists recent workflow sessions, optionally filtered by workflow_id, up to a specified count, using the Treasure Data client. Returns formatted session information including status and timing.
async def td_list_sessions( workflow_id: str | None = None, count: int = 20 ) -> dict[str, Any]: """List recent workflow executions to monitor status and find failures. Shows recent scheduled runs (sessions) with their execution status. Filter by workflow ID to see history of a specific workflow, or leave empty for all. Common scenarios: - Check which workflows ran recently and their status - Find failed executions that need investigation - Monitor execution patterns for a specific workflow - Get session IDs for detailed analysis Returns list with workflow names, execution times, and success/failure status. """ client = _create_client(include_workflow=True) if isinstance(client, dict): return client try: sessions = client.get_sessions(workflow_id=workflow_id, last=count) session_list = [] for session in sessions: session_info = { "id": session.id, "workflow": session.workflow["name"], "project": session.project["name"], "session_time": session.session_time, "status": session.last_attempt.status, "success": session.last_attempt.success, } # Add duration if available if session.last_attempt.created_at and session.last_attempt.finished_at: session_info["created_at"] = session.last_attempt.created_at session_info["finished_at"] = session.last_attempt.finished_at session_list.append(session_info) result = { "sessions": session_list, "count": len(session_list), } if workflow_id: result["filtered_by_workflow"] = workflow_id return result except Exception as e: return _format_error_response(f"Failed to list sessions: {str(e)}") - td_mcp_server/execution_tools.py:26-31 (registration)Registration of the 'td_list_sessions' tool (and related execution tools) using the mcp.tool() decorator within the register_execution_tools function.
mcp.tool()(td_get_session) mcp.tool()(td_list_sessions) mcp.tool()(td_get_attempt) mcp.tool()(td_get_attempt_tasks) mcp.tool()(td_analyze_execution)