dune-mcp

Instructions

Get columns for a table. Costs Credits (runs SELECT * LIMIT 0).

Implementation Reference

• src/main.py:130-152 (handler)

  MCP tool handler for get_table_schema. Manages budget checks, calls the Dune service to fetch schema, formats columns list as string response.

  def get_table_schema(table_name: str) -> str:
      """
      Get columns for a table. Costs Credits (runs SELECT * LIMIT 0).
      """
      try:
          # 1. Check Budget (This counts as a query execution)
          budget_manager.check_can_execute_query(estimated_cost=1) # Assume 1 credit base cost
          
          # 2. Execute
          schema = dune_service.get_table_schema(table_name)
          
          # 3. Track Budget
          budget_manager.track_execution(cost=1) # Track the spend
          
          cols = schema.get("columns", [])
          col_strs = [f"- {c['name']} ({c['type']})" for c in cols]
          
          return f"Schema for '{table_name}':\n" + "\n".join(col_strs)
  
      except BudgetExceededError as e:
          return f"SCHEMA ACCESS DENIED: {str(e)}"
      except Exception as e:
          return f"Error fetching schema: {str(e)}"

• src/services/dune_client.py:300-356 (helper)

  Core implementation in DuneClient service. Executes 'SELECT * FROM table LIMIT 0' via client.run_sql to get metadata, extracts column names and types.

  def get_table_schema(self, table_name: str) -> Dict[str, Any]:
      """
      Fetches the schema (columns and types) for a table by running a
      'SELECT * FROM table LIMIT 0' query.
      WARNING: This consumes Dune credits!
      """
      sql = f"SELECT * FROM {table_name} LIMIT 0"
      
      # We need to execute and wait for the result to get metadata
      # create_query usually requires a name, but we can use execute_query with raw sql 
      # via the query_id (if we had one) or generic execute.
      # dune-client 1.x allows executing raw SQL via `execute_query` if we pass a Query object 
      # but usually we need a query ID.
      
      # Actually, the official SDK/API usually requires a Query ID to execute anything.
      # We can't just send raw SQL without an existing Query ID container unless we create one.
      # BUT, we can use the "Query ID 0" or "Ad-hoc" mode if supported, or we must create a query.
      
      # Let's check if we can use `client.run_sql` or similar (from our introspection earlier).
      # We saw `run_sql`. Let's try that.
      
      try:
          # run_sql takes (query_sql, performance=...)
          # It returns a ResultsResponse
          result = self.client.run_sql(
              query_sql=sql,
              performance="medium" # medium is usually fine/cheapest
          )
          
          # The result object has 'meta' -> 'columns'
          # We need to inspect the structure of 'result'
          # Usually result.result.metadata.column_names / column_types
          
          if not result or not result.result:
              return {"error": "No result returned"}
  
          meta = result.result.metadata
          columns = []
          if meta:
              # Based on debugging, ResultMetadata has column_names and column_types directly
              if hasattr(meta, 'column_names') and hasattr(meta, 'column_types'):
                  for i, name in enumerate(meta.column_names):
                      # Ensure meta.column_types is also indexed by i
                      col_type = meta.column_types[i] if i < len(meta.column_types) else "unknown"
                      columns.append({"name": name, "type": col_type})
              else:
                  logger.warning("Could not find column_names/column_types in ResultMetadata.")
          
          return {
              "table": table_name,
              "columns": columns
          }
          
      except Exception as e:
          logger.error(f"Error getting schema for {table_name}: {e}")
          raise