td_get_session
Retrieve workflow session details by ID to verify execution status, schedule timing, and attempt information for scheduled Treasure Data workflow runs.
Instructions
Get workflow session details by ID to check execution status and timing.
A session is a scheduled workflow run. Use when you have a session ID and need
to check if it ran successfully, when it was scheduled, or get attempt details.
Common scenarios:
- Verify if a scheduled workflow executed at the expected time
- Get the attempt ID to investigate execution details
- Check overall success/failure status
Returns session info with workflow name, schedule time, and latest attempt status.Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| session_id | Yes |
Implementation Reference
- td_mcp_server/execution_tools.py:33-81 (handler)The main handler function for the 'td_get_session' tool. It takes a session_id parameter, fetches the session details from the Treasure Data client, formats the response with session and last_attempt information, and handles errors.
async def td_get_session(session_id: str) -> dict[str, Any]: """Get workflow session details by ID to check execution status and timing. A session is a scheduled workflow run. Use when you have a session ID and need to check if it ran successfully, when it was scheduled, or get attempt details. Common scenarios: - Verify if a scheduled workflow executed at the expected time - Get the attempt ID to investigate execution details - Check overall success/failure status Returns session info with workflow name, schedule time, and latest attempt status. """ if not session_id or not session_id.strip(): return _format_error_response("Session ID cannot be empty") client = _create_client(include_workflow=True) if isinstance(client, dict): return client try: session = client.get_session(session_id) if session: result = { "type": "session", "session": { "id": session.id, "project": session.project, "workflow": session.workflow, "session_uuid": session.session_uuid, "session_time": session.session_time, "last_attempt": { "id": session.last_attempt.id, "done": session.last_attempt.done, "success": session.last_attempt.success, "status": session.last_attempt.status, "created_at": session.last_attempt.created_at, "finished_at": session.last_attempt.finished_at, }, }, } return result else: return _format_error_response(f"Session with ID '{session_id}' not found") except Exception as e: return _format_error_response(f"Failed to get session: {str(e)}") - td_mcp_server/execution_tools.py:25-31 (registration)Registration of the 'td_get_session' tool (along with related execution tools) using the mcp.tool() decorator within the register_execution_tools function.
# Register all tools 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)