dune-mcp

Instructions

Implementation Reference

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

  MCP tool handler for 'get_table_schema'. Decorated with @mcp.tool(), handles budget checks, delegates to dune_service for schema retrieval, formats and returns column list as string.
  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:290-346 (helper)

  DuneClient service method that implements the core logic: executes 'SELECT * FROM table LIMIT 0' via Dune API to retrieve column names and types, returning a dict with 'columns' list.
  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