dune-mcp

Instructions

Implementation Reference

• src/main.py:187-209 (handler)

  The primary MCP tool handler for 'execute_query'. Decorated with @mcp.tool(), it checks session budget, delegates execution to DuneService, tracks usage, and returns the job ID with instructions.

  @mcp.tool()
  def execute_query(query_id: int, params: Optional[Dict[str, Any]] = None) -> str:
      """
      Execute a query by ID. Returns Job ID.
      """
      try:
          # 1. Check Session Budget (Query Count)
          # We don't know exact cost yet, so we pass 0 for now, or an estimate if we had one.
          budget_manager.check_can_execute_query(estimated_cost=0) 
          
          # 2. Execute
          job_id = dune_service.execute_query(query_id, params)
          
          # 3. Track (Assume minimal cost for now until we get result metadata)
          budget_manager.track_execution(cost=0) 
          
          return f"Execution started. Job ID: {job_id}. Use 'get_job_status' to check progress."
      
      except BudgetExceededError as e:
          return f"EXECUTION DENIED: {str(e)}"
      except Exception as e:
          return f"Error executing query: {str(e)}"

• src/services/dune_client.py:464-494 (helper)

  Supporting service method in DuneService that performs the actual API POST to /query/{query_id}/execute, returning the execution_id (job ID). Called by the MCP handler.

  def execute_query(self, query_id: int, params: Optional[Dict[str, Any]] = None) -> str:
      # We want to start execution and return ID, NOT wait.
      # client.execute_query() waits.
      # client.execute() (base method) usually returns the response with job_id.
      
      query = QueryBase(
          query_id=query_id, 
          params=[QueryParameter.from_dict(p) for p in (params or [])]
      )
      
      # Using the lower-level execute method to get job_id without waiting
      # The SDK implementation of execute() typically performs the POST /execute call
      try:
          # We construct the execution payload manually if SDK doesn't expose non-blocking nicely
          # Or checking SDK: client.execute(query) -> ExecutionResult (which contains job_id)
          # wait... client.execute() waits for completion loop.
          
          # Use raw request for async trigger to be safe and efficient
          url = f"{self.base_url}/query/{query_id}/execute"
          payload = {"query_parameters": {p["name"]: p["value"] for p in (params or [])}}
          headers = {"X-Dune-Api-Key": self.api_key}
          
          response = requests.post(url, json=payload, headers=headers)
          response.raise_for_status()
          data = response.json()
          return data["execution_id"]
          
      except Exception as e:
          logger.error(f"Error executing query {query_id}: {e}")
          raise

• src/services/budget_manager.py:23-30 (helper)

  BudgetManager helper method used by the handler to check if a query execution is allowed based on session limits before proceeding.

  def check_can_execute_query(self, estimated_cost: float = 0.0) -> None:
      """
      Check if a new query can be executed within the session budget.
      Raises BudgetExceededError if limits would be breached.
      """
      # Check query count limit
      if self.query_count >= self.config.max_queries:
          raise BudgetExceededError(