dune_query_create
Create and save SQL queries for blockchain data analysis on Dune Analytics. Define query names, SQL statements, descriptions, tags, and parameters for organized data exploration.
Instructions
Create a new saved Dune query (name + SQL). Requires SPICE_DUNE_ALLOW_SAVES=true.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| name | Yes | ||
| query_sql | Yes | ||
| description | No | ||
| tags | No | ||
| parameters | No |
Implementation Reference
- src/spice_mcp/mcp/server.py:794-799 (registration)Registration of the dune_query_create tool using @app.tool decorator with name, title, description, and tags.@app.tool( name="dune_query_create", title="Create Saved Query", description="Create a new saved Dune query (name + SQL). Requires SPICE_DUNE_ALLOW_SAVES=true.", tags={"dune", "admin"}, )
- src/spice_mcp/mcp/server.py:800-840 (handler)The main execution handler for dune_query_create tool, validates config, delegates to QUERY_ADMIN_SERVICE.create, logs success/error.def dune_query_create(name: str, query_sql: str, description: str | None = None, tags: list[str] | None = None, parameters: list[dict[str, Any]] | None = None) -> dict[str, Any]: _ensure_initialized() if CONFIG is None or not CONFIG.allow_saves: return error_response( ValueError("Saving queries is disabled. Set SPICE_DUNE_ALLOW_SAVES=true to enable."), context={"tool": "dune_query_create", "allow_saves": False} ) assert QUERY_ADMIN_SERVICE is not None try: result = dict(QUERY_ADMIN_SERVICE.create(name=name, query_sql=query_sql, description=description, tags=tags, parameters=parameters)) # Log admin action if QUERY_HISTORY is not None: query_id = result.get("query_id") if query_id: QUERY_HISTORY.record( execution_id=f"create_{query_id}", query_type="query_id", query_preview=f"Created query: {name}", status="success", duration_ms=0, action_type="admin_action", query_id=query_id, action="create", name=name, ) return result except Exception as e: # Log error if QUERY_HISTORY is not None: QUERY_HISTORY.record( execution_id=f"create_failed", query_type="raw_sql", query_preview=f"Failed to create query: {name}", status="error", duration_ms=0, action_type="admin_action", action="create", name=name, error=str(e), ) return error_response(e, context={"tool": "dune_query_create", "name": name})
- QueryAdminService.create method: wrapper that sets defaults for is_private and delegates to admin port.def create(self, *, name: str, query_sql: str, description: str | None = None, tags: Sequence[str] | None = None, parameters: Sequence[Mapping[str, Any]] | None = None, is_private: bool | None = None) -> Mapping[str, Any]: # Apply force_private override if enabled if self.force_private: is_private = True elif is_private is None: is_private = False return self.admin.create(name=name, query_sql=query_sql, description=description, tags=list(tags) if tags else None, parameters=list(parameters) if parameters else None, is_private=is_private)
- Core implementation in DuneAdminAdapter.create: constructs JSON body and makes POST request to Dune API /api/v1/query/create to create the saved query.def create( self, *, name: str, query_sql: str, description: str | None = None, tags: list[str] | None = None, parameters: list[dict] | None = None, is_private: bool = False, ) -> dict[str, Any]: body: dict[str, Any] = {"name": name, "query_sql": query_sql, "is_private": is_private} if description: body["description"] = description # Auto-attach 'spice-mcp' tag if tags not provided if tags is None: body["tags"] = ["spice-mcp"] else: body["tags"] = tags if parameters is not None: body["parameters"] = parameters resp = self._http.request( "POST", urls.url_templates["query_create"], headers=self._headers(), json=body, timeout=20, ) return resp.json()