td_get_attempt
Retrieve workflow execution details to investigate specific run instances, including status, timing, retry information, and parameters for debugging purposes.
Instructions
Get workflow attempt details to investigate specific execution instance.
An attempt is one execution try of a scheduled session. Use when you have an
attempt ID from error logs or td_get_session and need execution details.
Common scenarios:
- Investigate why a workflow execution failed
- Check how long the execution took
- See if this was a retry after previous failure
- Get execution parameters for debugging
Returns attempt status, timing, retry info, and safe execution parameters.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| attempt_id | Yes |
Implementation Reference
- td_mcp_server/execution_tools.py:138-217 (handler)The handler function implementing the td_get_attempt tool. It retrieves workflow attempt details from the Treasure Data client, calculates duration, filters sensitive params, and handles errors.async def td_get_attempt(attempt_id: str) -> dict[str, Any]: """Get workflow attempt details to investigate specific execution instance. An attempt is one execution try of a scheduled session. Use when you have an attempt ID from error logs or td_get_session and need execution details. Common scenarios: - Investigate why a workflow execution failed - Check how long the execution took - See if this was a retry after previous failure - Get execution parameters for debugging Returns attempt status, timing, retry info, and safe execution parameters. """ if not attempt_id or not attempt_id.strip(): return _format_error_response("Attempt ID cannot be empty") client = _create_client(include_workflow=True) if isinstance(client, dict): return client try: attempt = client.get_attempt(attempt_id) if attempt: result = { "type": "attempt", "attempt": { "id": attempt.id, "index": attempt.index, "project": attempt.project, "workflow": attempt.workflow, "session_id": attempt.session_id, "session_time": attempt.session_time, "retry_attempt_name": attempt.retry_attempt_name, "done": attempt.done, "success": attempt.success, "status": attempt.status, "created_at": attempt.created_at, "finished_at": attempt.finished_at, }, } # Add duration if finished if attempt.created_at and attempt.finished_at: try: from datetime import datetime created = datetime.fromisoformat( attempt.created_at.replace("Z", "+00:00") ) finished = datetime.fromisoformat( attempt.finished_at.replace("Z", "+00:00") ) duration = finished - created result["attempt"]["duration_seconds"] = duration.total_seconds() except Exception: pass # Ignore date parsing errors # Add non-sensitive parameters if attempt.params: # Filter out sensitive parameters safe_params = { k: v for k, v in attempt.params.items() if not any( sensitive in k.lower() for sensitive in ["email", "ip", "user_id", "token", "key"] ) } if safe_params: result["attempt"]["params"] = safe_params return result else: return _format_error_response(f"Attempt with ID '{attempt_id}' not found") except Exception as e: return _format_error_response(f"Failed to get attempt: {str(e)}")
- td_mcp_server/execution_tools.py:26-30 (registration)Registration of execution tools including td_get_attempt 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)