get_job_status
Check the execution status of a Dune Analytics query job by providing the job ID. Monitors progress for up to 30 seconds to confirm completion.
Instructions
Check query execution status. Polls for 30s.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| job_id | Yes |
Implementation Reference
- src/main.py:210-238 (handler)The main handler implementation for the 'get_job_status' MCP tool. It polls the Dune service for the job status up to 30 seconds (15 retries of 2s each), reports completion/failure/cancellation with credits used if available, or indicates if still running.@mcp.tool() def get_job_status(job_id: str) -> str: """ Check query execution status. Polls for 30s. """ try: # Internal polling loop (max 30s) import time max_retries = 15 # 15 * 2s = 30s for i in range(max_retries): status_data = dune_service.get_status(job_id) state = status_data.get("state", "UNKNOWN") if state in ["COMPLETED", "FAILED", "CANCELLED"]: msg = f"Job {job_id} is {state}" credits = status_data.get("credits_used") if credits is not None: msg += f" (Cost: {credits} Credits)" return msg # If running, wait a bit time.sleep(2) # If still running after 30s return f"Job {job_id} is still {state}. Please check again later." except Exception as e: return f"Error checking status: {str(e)}"