Treasure Data MCP Server

Instructions

Implementation Reference

• td_mcp_server/url_tools.py:93-213 (handler)

  The main handler function for the td_get_workflow tool. Fetches workflow details by numeric ID using the Treasure Data client API, with direct lookup and fallback to searching all workflows. Includes project info, schedule, recent sessions, and console URL.
  async def td_get_workflow(workflow_id: str) -> dict[str, Any]: """Get workflow details using numeric ID - essential for console URLs. Direct workflow lookup when you have the ID. Handles large workflow IDs that exceed pagination limits. Returns project info and execution history. Common scenarios: - Extracting ID from console URL (../workflows/12345678/info) - Looking up workflow from error logs containing ID - Getting project context for a known workflow ID - Checking execution status by workflow ID Returns workflow name, project details, schedule, and recent runs. Includes console URL for quick browser access. """ if not workflow_id or not workflow_id.strip(): return _format_error_response("Workflow ID cannot be empty") # Validate workflow ID format if not re.match(r"^\d+$", workflow_id): return _format_error_response("Invalid workflow ID format. Must be numeric.") client = _create_client(include_workflow=True) if isinstance(client, dict): return client try: # First try the direct API endpoint workflow = client.get_workflow_by_id(workflow_id) if workflow: # Found the workflow via direct API result: dict[str, Any] = { "type": "workflow", "workflow": { "id": workflow.id, "name": workflow.name, "project": { "id": workflow.project.id, "name": workflow.project.name, }, "timezone": workflow.timezone, "scheduled": workflow.schedule is not None, }, } # Add schedule info if available if workflow.schedule: result["workflow"]["schedule"] = workflow.schedule # Add latest session info if available # Note: Direct API might not include session info if workflow.latest_sessions: latest_sessions = [] for session in workflow.latest_sessions[:5]: # Last 5 sessions latest_sessions.append( { "session_time": session.session_time, "status": session.last_attempt.status, "success": session.last_attempt.success, } ) result["workflow"]["latest_sessions"] = latest_sessions # Construct console URL result[ "console_url" ] = f"https://console.treasuredata.com/app/workflows/{workflow_id}/info" return result # If not found via direct API, fall back to searching through all workflows # This might be needed for workflows accessible via console API only workflows = client.get_workflows(count=1000, all_results=True) for workflow in workflows: if workflow.id == workflow_id: # Found the workflow result = { "type": "workflow", "workflow": { "id": workflow.id, "name": workflow.name, "project": { "id": workflow.project.id, "name": workflow.project.name, }, "timezone": workflow.timezone, "scheduled": workflow.schedule is not None, }, } # Add schedule info if available if workflow.schedule: result["workflow"]["schedule"] = workflow.schedule # Add latest session info if available if workflow.latest_sessions: latest_sessions = [] for session in workflow.latest_sessions[:5]: # Last 5 sessions latest_sessions.append( { "session_time": session.session_time, "status": session.last_attempt.status, "success": session.last_attempt.success, } ) result["workflow"]["latest_sessions"] = latest_sessions # Construct console URL result[ "console_url" ] = f"https://console.treasuredata.com/app/workflows/{workflow_id}/info" return result return _format_error_response(f"Workflow with ID '{workflow_id}' not found") except Exception as e: return _format_error_response(f"Failed to get workflow: {str(e)}")
• td_mcp_server/url_tools.py:28-28 (registration)

  Registers the td_get_workflow tool with the MCP instance using the mcp.tool() decorator.
  mcp.tool()(td_get_workflow)