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})

src/spice_mcp/service_layer/query_admin_service.py:19-25 (helper)

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)

src/spice_mcp/adapters/dune/admin.py:36-63 (helper)

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()

Spice MCP