dune-mcp

Instructions

Implementation Reference

• src/main.py:352-365 (handler)

  MCP tool handler and registration for 'search_spellbook'. Thin wrapper that calls DuneService.search_spellbook(keyword) and formats the list of matching Spellbook files (SQL models and schema.yml) into a string summary.
  @mcp.tool() def search_spellbook(keyword: str) -> str: """ Search GitHub Spellbook for official tables (.sql/.yml). Zero-credit schema discovery. """ results = dune_service.search_spellbook(keyword) if not results: return f"No results found in Spellbook for '{keyword}'." summary = [] for f in results[:15]: # Limit to top 15 matches summary.append(f"[{f['type']}] {f['path']}") return "\n".join(summary)
• src/services/dune_client.py:67-114 (helper)

  Core implementation of search_spellbook in DuneService class. Performs GitHub code search on duneanalytics/spellbook repo for keyword in .sql files and schema.yml files. Returns list of matching file dicts with path, type (sql_model/schema_definition), etc. Uses caching.
  def search_spellbook(self, keyword: str) -> List[Dict[str, Any]]: """ Searches the duneanalytics/spellbook GitHub repository for SQL models and schema files. Results are cached for 24 hours. """ cache_key = f"search:{keyword}" cached_results = self.cache.get("github", cache_key) if cached_results: return cached_results base_url = "https://api.github.com/search/code" repo = "repo:duneanalytics/spellbook" # Search for .sql files sql_query = f"{keyword} {repo} in:file extension:sql" sql_url = f"{base_url}?q={sql_query}" sql_results = self._github_api_request(sql_url) # Search for schema.yml files (which might contain definitions) yaml_query = f"{keyword} {repo} in:file filename:schema.yml" yaml_url = f"{base_url}?q={yaml_query}" yaml_results = self._github_api_request(yaml_url) found_files = [] if sql_results and sql_results.get("items"): for item in sql_results["items"]: found_files.append({ "name": item["name"], "path": item["path"], "url": item["html_url"], "type": "sql_model", "repo_url": item["url"] # API URL for file content }) if yaml_results and yaml_results.get("items"): for item in yaml_results["items"]: found_files.append({ "name": item["name"], "path": item["path"], "url": item["html_url"], "type": "schema_definition", "repo_url": item["url"] }) if found_files: self.cache.set("github", cache_key, found_files) return found_files