dune-mcp

Instructions

Implementation Reference

• src/main.py:211-237 (handler)

  The main handler function for the 'get_job_status' tool, decorated with @mcp.tool() for registration. It polls the Dune service for job status up to 30 seconds and returns the status or error.

  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)}"