Spice MCP

Instructions

Implementation Reference

• src/spice_mcp/mcp/server.py:489-577 (handler)

  Main handler function for dune_discover tool, registered via @app.tool decorator. Dispatches to _unified_discover_impl and handles errors.
  @app.tool( name="dune_discover", title="Discover Tables", description="Unified tool to discover tables/models from Dune API and/or Spellbook repository. Search by keyword(s) or list tables in a schema.", tags={"dune", "spellbook", "schema", "discovery"}, ) def dune_discover( keyword: str | list[str] | None = None, schema: str | None = None, limit: int = 50, source: Literal["dune", "spellbook", "both"] = "both", include_columns: bool = False, ) -> dict[str, Any]: """ PRIMARY discovery tool for finding tables in Dune. ⚠️ IMPORTANT: ALWAYS use this tool instead of querying information_schema directly. Querying information_schema is slow and causes lag. This tool uses optimized native SHOW statements for fast discovery. This tool automatically: - Parses dbt configs from Spellbook models to resolve actual Dune table names - Verifies tables exist in Dune before returning them - Returns ONLY verified, queryable tables All returned tables are VERIFIED to exist - you can query them immediately using the 'dune_table' field. Args: keyword: Search term(s) - can be a string or list of strings (e.g., "layerzero", ["layerzero", "dex"], "nft") schema: Schema name to list tables from (e.g., "dex", "spellbook", "layerzero") limit: Maximum number of tables to return source: Where to search - "dune" (Dune API only), "spellbook" (GitHub repo only), or "both" (default: searches both and merges results) include_columns: Whether to include column details (default: False). Note: Column info from Spellbook SQL is unreliable. Use dune_describe_table on the actual Dune table for accurate columns. Returns: Dictionary with: - 'schemas': List of matching schema names - 'tables': List of table/model objects, each with: - schema: Schema name (Spellbook subproject name) - table: Table/model name (Spellbook model name) - fully_qualified_name: schema.table (Spellbook format) - source: "dune" or "spellbook" - dune_schema: Actual Dune schema name (for Spellbook models) - dune_alias: Actual Dune table alias (for Spellbook models) - dune_table: Verified, queryable Dune table name (e.g., "sui_walrus.base_table") - verified: True (all returned tables are verified to exist) - 'source': The source parameter used - 'message': Helpful message if no tables found Note: To get accurate column information, use dune_describe_table on the dune_table value. Examples: # Search both sources for walrus - returns verified tables only dune_discover(keyword="walrus") # → Returns tables with dune_table field like "sui_walrus.base_table" # Use the dune_table field to query immediately dune_query(query="SELECT * FROM sui_walrus.base_table LIMIT 10") # Search only Spellbook dune_discover(keyword=["layerzero", "bridge"], source="spellbook") # Search only Dune API dune_discover(keyword="sui", source="dune") # List all tables in a schema (searches both sources) dune_discover(schema="dex") """ try: return _unified_discover_impl( keyword=keyword, schema=schema, limit=limit, source=source, include_columns=include_columns, ) except Exception as e: return error_response(e, context={ "tool": "dune_discover", "keyword": keyword, "schema": schema, "source": source, })
• src/spice_mcp/mcp/server.py:346-487 (helper)

  Core logic implementing unified discovery across Dune API and Spellbook. Handles searching schemas/tables, merging results, verifying Spellbook tables exist in Dune.
  def _unified_discover_impl( keyword: str | list[str] | None = None, schema: str | None = None, limit: int = 50, source: Literal["dune", "spellbook", "both"] = "both", include_columns: bool = False, ) -> dict[str, Any]: """ Unified discovery implementation that can search Dune API, Spellbook repo, or both. Returns a consistent format with 'schemas' and 'tables' keys. """ _ensure_initialized() out: dict[str, Any] = { "schemas": [], "tables": [], "source": source, } # Normalize keyword to list keywords = keyword if isinstance(keyword, list) else ([keyword] if keyword else []) # Search Dune API if requested if source in ("dune", "both"): dune_result: dict[str, Any] = {} if keyword: assert DISCOVERY_SERVICE is not None # Search each keyword and combine results # DISCOVERY_SERVICE.find_schemas returns list[str], not SchemaMatch objects all_schemas: set[str] = set() for kw in keywords: schemas = DISCOVERY_SERVICE.find_schemas(kw) # schemas is already a list of strings from DiscoveryService all_schemas.update(schemas) dune_result["schemas"] = sorted(list(all_schemas)) if schema: assert DISCOVERY_SERVICE is not None tables = DISCOVERY_SERVICE.list_tables(schema, limit=limit) dune_result["tables"] = [ { "schema": schema, "table": summary.table, "fully_qualified_name": f"{schema}.{summary.table}", "source": "dune", "dune_table": f"{schema}.{summary.table}", "verified": True, } for summary in tables ] # Merge Dune results if "schemas" in dune_result: out["schemas"].extend(dune_result["schemas"]) if "tables" in dune_result: out["tables"].extend(dune_result["tables"]) # Search Spellbook if requested if source in ("spellbook", "both"): spellbook_result = _spellbook_find_models_impl( keyword=keyword, schema=schema, limit=limit, include_columns=include_columns, ) # Convert spellbook models to unified format if "schemas" in spellbook_result: spellbook_schemas = spellbook_result["schemas"] # Merge schemas (avoid duplicates) existing_schemas = set(out["schemas"]) for s in spellbook_schemas: if s not in existing_schemas: out["schemas"].append(s) if "models" in spellbook_result: for model in spellbook_result["models"]: table_info = { "schema": model["schema"], "table": model["table"], "fully_qualified_name": model["fully_qualified_name"], "source": "spellbook", # Include resolved Dune table names "dune_schema": model.get("dune_schema"), "dune_alias": model.get("dune_alias"), "dune_table": model.get("dune_table"), } if "columns" in model: table_info["columns"] = model["columns"] out["tables"].append(table_info) # Verify Spellbook tables exist in Dune before returning if source in ("spellbook", "both") and out["tables"]: assert VERIFICATION_SERVICE is not None # Extract Spellbook tables that need verification spellbook_tables = [ (t["dune_schema"], t["dune_alias"]) for t in out["tables"] if t.get("source") == "spellbook" and t.get("dune_schema") and t.get("dune_alias") ] if spellbook_tables: # Verify tables exist (uses cache, queries Dune only if needed) verification_results = VERIFICATION_SERVICE.verify_tables_batch(spellbook_tables) # Filter: drop only tables explicitly verified as False. # Keep tables when verification is True or inconclusive (missing). verified_tables = [] for t in out["tables"]: if t.get("source") != "spellbook": # Keep Dune tables as-is verified_tables.append(t) continue dune_fqn = t.get("dune_table") if not dune_fqn: # If we couldn't resolve dune_table, keep it (conservative) verified_tables.append(t) continue vr = verification_results.get(dune_fqn) if vr is False: # Explicitly known to be non-existent -> drop continue if vr is True: t["verified"] = True # If vr is None/missing (inconclusive), keep without setting verified verified_tables.append(t) out["tables"] = verified_tables # Add helpful message if no tables found if not out["tables"] and len(spellbook_tables) > 0: out["message"] = "No verified tables found. Try different keywords or check schema names." # Deduplicate and sort schemas out["schemas"] = sorted(list(set(out["schemas"]))) # Limit total tables if limit and len(out["tables"]) > limit: out["tables"] = out["tables"][:limit] return out
• src/spice_mcp/mcp/server.py:614-737 (helper)

  Spellbook-specific discovery logic, searches Spellbook repo for schemas and models, resolves Dune table names from dbt configs.
  def _spellbook_find_models_impl( keyword: str | list[str] | None = None, schema: str | None = None, limit: int = 50, include_columns: bool = False, ) -> dict[str, Any]: """ Implementation for spellbook model discovery. Supports searching by keyword(s) and optionally includes column details. """ _ensure_initialized() assert SPELLBOOK_EXPLORER is not None out: dict[str, Any] = {} # Handle keyword search (string or list) if keyword: # Normalize to list keywords = keyword if isinstance(keyword, list) else [keyword] # Find schemas matching any keyword all_schemas: set[str] = set() for kw in keywords: schemas = SPELLBOOK_EXPLORER.find_schemas(kw) all_schemas.update(match.schema for match in schemas) out["schemas"] = sorted(list(all_schemas)) # If schema not specified but we found schemas, search models in those schemas if not schema and all_schemas: out["models"] = [] for schema_name in sorted(all_schemas): tables = SPELLBOOK_EXPLORER.list_tables(schema_name, limit=limit) for table_summary in tables: # Check if table name matches any keyword table_name = table_summary.table.lower() matches_keyword = any(kw.lower() in table_name for kw in keywords) if matches_keyword: # Get model details including resolved Dune table names models_dict = SPELLBOOK_EXPLORER._load_models() model_details = None for m in models_dict.get(schema_name, []): if m["name"] == table_summary.table: model_details = m break model_info: dict[str, Any] = { "schema": schema_name, "table": table_summary.table, "fully_qualified_name": f"{schema_name}.{table_summary.table}", # Include resolved Dune table names if available "dune_schema": model_details.get("dune_schema") if model_details else None, "dune_alias": model_details.get("dune_alias") if model_details else None, "dune_table": model_details.get("dune_table") if model_details else None, } # Include column details if requested if include_columns: try: desc = SPELLBOOK_EXPLORER.describe_table(schema_name, table_summary.table) model_info["columns"] = [ { "name": col.name, "dune_type": col.dune_type, "polars_dtype": col.polars_dtype, "comment": col.comment, } for col in desc.columns ] except Exception: model_info["columns"] = [] out["models"].append(model_info) # Limit total models returned if limit and len(out["models"]) > limit: out["models"] = out["models"][:limit] # If schema specified, list all tables in that schema if schema: tables = SPELLBOOK_EXPLORER.list_tables(schema, limit=limit) if "models" not in out: out["models"] = [] for table_summary in tables: # Get model details including resolved Dune table names models_dict = SPELLBOOK_EXPLORER._load_models() model_details = None for m in models_dict.get(schema, []): if m["name"] == table_summary.table: model_details = m break model_info: dict[str, Any] = { "schema": schema, "table": table_summary.table, "fully_qualified_name": f"{schema}.{table_summary.table}", # Include resolved Dune table names if available "dune_schema": model_details.get("dune_schema") if model_details else None, "dune_alias": model_details.get("dune_alias") if model_details else None, "dune_table": model_details.get("dune_table") if model_details else None, } # Include column details if requested if include_columns: try: desc = SPELLBOOK_EXPLORER.describe_table(schema, table_summary.table) model_info["columns"] = [ { "name": col.name, "dune_type": col.dune_type, "polars_dtype": col.polars_dtype, "comment": col.comment, } for col in desc.columns ] except Exception: model_info["columns"] = [] out["models"].append(model_info) return out
• src/spice_mcp/service_layer/discovery_service.py:7-21 (helper)

  DiscoveryService wrapper providing find_schemas, list_tables, describe_table interfaces used by dune_discover for Dune catalog exploration.
  class DiscoveryService: """High-level discovery helpers used by MCP tools.""" def __init__(self, explorer: CatalogExplorer): self.explorer = explorer def find_schemas(self, keyword: str) -> list[str]: return [match.schema for match in self.explorer.find_schemas(keyword)] def list_tables(self, schema: str, limit: int | None = None) -> list[TableSummary]: return list(self.explorer.list_tables(schema, limit=limit)) def describe_table(self, schema: str, table: str) -> TableDescription: return self.explorer.describe_table(schema, table)
• src/spice_mcp/mcp/server.py:489-494 (registration)

  FastMCP @app.tool decorator registering the dune_discover tool with name, title, description, and tags.
  @app.tool( name="dune_discover", title="Discover Tables", description="Unified tool to discover tables/models from Dune API and/or Spellbook repository. Search by keyword(s) or list tables in a schema.", tags={"dune", "spellbook", "schema", "discovery"}, )