semql-mcp

An MCP server that wraps a semql Catalog and exposes its compiler / validator / prompt-renderer surfaces as tools any MCP client can call. Built on FastMCP.

Two modes

By default the server is compile-only. semql is a pure compiler — no I/O — and this server keeps that contract. Tools return the emitted SQL and bound parameters; the caller runs the SQL against whatever backend they own.

Pass an executor at construction to opt into exec mode. A query_execute tool registers in addition to the compile-only tools; it runs the SQL against your executor and returns both the SQL/params envelope and the resulting rows.

Install

pip install semql-mcp

Quick start — compile-only

from semql import Dialect, Catalog, Cube, Dimension, Measure
from semql_mcp import MCPServer

catalog = Catalog([
    Cube(
        name="orders",
        dialect=Dialect.POSTGRES,
        table="orders",
        alias="o",
        measures=[Measure(name="revenue", sql="{o}.amount", agg="sum", unit="currency")],
        dimensions=[Dimension(name="region", sql="{o}.region", type="string")],
    ),
])

server = MCPServer(catalog)
server.run(transport="stdio")  # speak JSON-RPC over stdin/stdout

Quick start — exec mode

Bring your own database driver and adapt its row shape to a list of dicts:

import psycopg
from psycopg.rows import dict_row

from semql_mcp import MCPServer


def executor(sql: str, params: dict) -> list[dict]:
    with psycopg.connect("postgresql://...", row_factory=dict_row) as conn:
        with conn.cursor() as cur:
            cur.execute(sql, params)
            return list(cur.fetchall())


server = MCPServer(catalog, executor=executor)
server.run(transport="stdio")

The MCP server never imports a database driver. Whatever you wire in is what gets called; semql-mcp just hands it (sql, params) and expects list[dict] back.

Tools

Always registered:

Registered when executor is supplied:

Auto-generated per-cube tools

For each expose_in_prompt=True (non-META) cube, the server also registers a query_<cube_name> tool whose measures, dimensions, order (and time_window.dimension, when applicable) parameters are Literal-typed enums of the cube's actual fields. The planner sees a JSON Schema with explicit allowed values rather than the bare list[str] query_semantic accepts.

Field names are bare (no cube prefix); the tool auto-qualifies as it builds the SemanticQuery:

// query_orders
{
  "measures": ["revenue"],
  "dimensions": ["region"],
  "filters": [{"dimension": "status", "op": "eq", "values": ["paid"]}],
  "time_window": {
    "dimension": "created_at",
    "granularity": "day",
    "range": ["2026-01-01", "2026-02-01"]
  },
  "limit": 100
}

Multi-cube queries (joins across cubes) still go through query_semantic — the per-cube tools are scoped to a single cube by construction. When executor is configured, the per-cube tools return rows too.

Authorization and metadata disclosure

Per-call execution is authorized and fails closed. Pass a viewer_provider and every tool — query_*, validate, catalog_prompt, lookup, and entity tools — resolves the viewer and refuses cubes, fields, lookups, and saved queries it isn't allowed to see. An unauthorized call returns a structured error, never data or SQL.

Accepted limitation — tool listing is not viewer-filtered. Tools are registered once when the server is constructed, before any client connects, so there is no request identity at registration time. A client that lists tools therefore sees the names, descriptions, and field enums of every role-gated cube, saved query, and entity in the catalog — including ones it cannot execute. This exposes catalog structure (names and shapes), never row data or SQL, and execution remains authorized. Treat tool names and descriptions as non-secret: don't encode confidential facts in cube / field / saved-query names if untrusted clients can list tools. To gate listing itself, run a per-tenant server (one catalog per trust boundary) or front it with a transport that filters the advertised tool set per connection.

In-process testing

FastMCP's Client connects to a FastMCP instance without a transport — useful for end-to-end testing of your catalog + planner together:

import asyncio
from fastmcp import Client
from semql_mcp import MCPServer

server = MCPServer(catalog)

async def smoke() -> None:
    async with Client(server.mcp) as c:
        tools = await c.list_tools()
        print([t.name for t in tools])
        result = await c.call_tool("explain", {"spec": {"measures": ["orders.revenue"]}})
        print(result.data)

asyncio.run(smoke())

Status

Early development. The tool surface is stable.