akko-mcp-trino

English · Français

Give your AI agents read-only access to Trino, and nothing more.

akko-mcp-trino is an MCP server for Trino. It lets an agent list catalogs, schemas and tables, describe a table, explain a query and run it, and it guarantees that nothing it sends can write: every statement is parsed, and a write anywhere in it is refused before Trino. Every query runs either as the one Trino identity the administrator configured, or as the user the agent acts for, whose own token Trino verifies again. Either way Trino's access control (OPA, Ranger, or Trino's own rules) decides what the agent may read. Results are bounded in rows and in time, and every call leaves one audit line.

Why

An agent that reads a lakehouse needs a small, predictable surface: a few tools whose descriptions a model understands, answers that never run away, and a guarantee that a prompt injection cannot turn a question into a DROP TABLE. This server is that surface, and nothing else. It does not decide access: in jwt mode Trino decides for each user with the rules it already has, and in service mode for the one identity you give it, with the rights every caller of this server may have, no more.

Highlights

• Read-only by construction. SQL is parsed into an AST and refused if a write appears anywhere in the tree, CTEs and subqueries included. Table functions (TABLE(...), such as a JDBC connector's system.query) and the system catalog are refused too, including through a name without a catalog when the session catalog is system.
• The user's own rights, end to end. In jwt mode the caller's verified token goes to Trino, which verifies it again and runs the query as that user: Ranger or OPA decide for the person, not for a service account, with no password and no right to act for another user. A request without a verified token is refused, never run as anyone else.
• Or one identity, configured by the administrator. In service mode, the default, every query runs as TRINO_USER, authenticated by TRINO_PASSWORD over https (or nothing, for a local Trino). The agent cannot name another user.
• Bounded. A row cap with a truncated flag, a time budget per query cancelled in Trino, a time budget per tool call.
• One audit line per call. Keyed by X-Request-Id, never the token, never the SQL text: who called which tool, with which outcome, for how long.
• Six tools, no more. List catalogs, schemas and tables, describe a table, explain a statement, run a read-only query.
• A guarded door. Bearer tokens verified against any OIDC provider's JWKS (issuer, audience, expiry, clock tolerance), strict mode refusing a request without a verified caller, a Host header filter.
• Built to be extended without being forked. Tool plugins and extensions are Python packages named at start: identity modes, guards at the door, audit fields, routes, tools. An extension can only refuse what the core allows, never widen it.
• Any MCP host, any model. Cursor, Claude Desktop, VS Code, the Python SDK… Mistral, or any OpenAI-compatible model through the example agent.
• Small and proven. 322 tests at 100 % line and branch coverage, including in-process suites on the real SDK for all three transports, and a product-neutral guard in CI.

Documentation

• How it works, end to end: akko-ai.com/en/docs/how-it-works (also en français).
• Guide, prerequisites, configuration, examples: akko-ai.com/docs/akko-mcp-trino (also in English) and akko-ai.com/docs/exemples.
• Reference, in this repository: this README, examples/, CHANGELOG.md, CONTRIBUTING.md, SECURITY.md.

Contents

1. How it works
2. Compatibility
3. Prerequisites
4. Install and run
5. Connect an MCP host
6. Use it from an agent
7. The tools
8. Extensions
9. What happens on a request
10. Configuration
11. Operations
12. Development
13. Design notes
14. Contributing
15. About AKKO

How it works

flowchart LR
    Host["MCP host<br/>(Cursor, Claude, VS Code, an agent)"]
    IdP["Identity provider<br/>(OIDC, JWKS)"]
    Server["akko-mcp-trino"]
    Trino["Trino"]
    Policy["Policy engine<br/>(OPA, Ranger, built-in)"]
    Data[("Data sources")]

    Host -- "1. tool call + Bearer token" --> Server
    Server -. "verify signature<br/>against JWKS" .-> IdP
    Server -- "2. read-only SQL, as the caller (jwt)<br/>or as TRINO_USER (service)" --> Trino
    Trino -- "3. may this identity read this?" --> Policy
    Trino -- "4. rows, as that identity may read them" --> Server
    Server -- "5. result" --> Host
    Trino --> Data

The host calls a tool. When authentication is on, the server verifies the caller's token (signature against the provider's JWKS, issuer, audience, expiry) and refuses a request without one in strict mode; the caller's name goes into the audit line. The server then checks the statement (read-only, identifiers, table functions, the system catalog) and runs it in Trino under the identity mode TRINO_IDENTITY_MODE names. Trino asks its policy engine and returns what that identity may read, bounded in rows and time.

In service mode the caller's token never leaves the server. In jwt mode it goes to Trino only once verified, over https only, and never into the audit line; without one, the query is refused. An extension (see Extensions) may add other identity modes.

Compatibility

Prerequisites

Optional: a Prometheus to scrape /metrics.

Install and run

pip install akko-mcp-trino        # or: pipx install akko-mcp-trino
uvx akko-mcp-trino --version      # run it without installing, the way most MCP hosts do

Or from source:

git clone https://github.com/AKKO-p/akko-mcp-trino.git
cd akko-mcp-trino
python -m venv .venv && source .venv/bin/activate
pip install -e .

The package installs a Python module (akko_mcp_trino) and a command (akko-mcp-trino); python -m akko_mcp_trino is the same entrypoint.

Point it at your Trino and your identity provider, then serve:

export TRINO_HOST=trino.example.internal
export TRINO_PORT=8443
export TRINO_HTTP_SCHEME=https
export TRINO_CATALOG=iceberg
export TRINO_USER=mcp-reader
export TRINO_PASSWORD=...               # Basic authentication, https only

export MCP_AUTH_ENABLED=true
export MCP_AUTH_REQUIRED=true          # refuse requests without a verified identity
export MCP_JWKS_URL=https://idp.example.com/realms/data/protocol/openid-connect/certs
export MCP_OIDC_ISSUER=https://idp.example.com/realms/data
export MCP_OIDC_AUDIENCE=data-platform

python -m akko_mcp_trino

Before serving, akko-mcp-trino --check prints the effective configuration (never a secret) and exits with 2 if something would refuse to start; over stdio it checks that MCP_USER_TOKEN is present, not that it is valid (the real start verifies the token). Then:

INFO:__main__:serving transport=streamable-http port=3000 auth=True strict=True extensions=none

Check it is alive and can reach Trino:

curl -s localhost:3001/health   # {"status":"ok"}       never touches Trino
curl -s localhost:3001/ready    # {"status":"ready"}    service: a bounded SELECT 1 as TRINO_USER; jwt: Trino's /v1/info
curl -s localhost:3001/metrics  # Prometheus exposition

Every query runs as TRINO_USER, whoever calls: give that identity the rights every caller of this server may have, no more. In strict mode, a request without a verified token is refused before any query.

Each user with their own rights (jwt)

When Trino trusts the same identity provider, let it decide for each user:

export TRINO_IDENTITY_MODE=jwt         # the caller's verified token goes to Trino
export TRINO_HTTP_SCHEME=https         # required: a token never travels over http
export MCP_AUTH_ENABLED=true
export MCP_AUTH_REQUIRED=true
# TRINO_USER and TRINO_PASSWORD are not used

On the Trino side, http-server.authentication.type=JWT, http-server.authentication.jwt.key-file set to the provider's JWKS URL, and http-server.authentication.jwt.principal-field naming the same claim the server reads (preferred_username or sub); required-issuer and required-audience pin the token to your provider and to Trino. Two users asking the same question then get the answers their own rights allow. /ready asks Trino's public /v1/info, since there is no identity of the server's own to query with.

Container

A Dockerfile ships with the repository (non-root, health check, OCI labels). Run it with the same environment variables, publishing ports 3000 (MCP) and 3001 (health).

Connect an MCP host

With authentication on, the host sends the user's access token from the identity provider; the server verifies it and names the caller in the audit line.

Cursor, Claude Desktop, VS Code (mcp.json):

{
  "mcpServers": {
    "trino": {
      "url": "http://localhost:3000/mcp",
      "headers": {"Authorization": "Bearer <the user's access token>"}
    }
  }
}

From Python, with the official SDK:

import anyio
from mcp import ClientSession
from mcp.client.streamable_http import create_mcp_http_client, streamable_http_client


async def main(token: str):
    http = create_mcp_http_client(headers={"Authorization": f"Bearer {token}"})
    url = "http://localhost:3000/mcp"
    async with streamable_http_client(url, http_client=http) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            result = await session.call_tool("execute_query", {"sql": "SELECT 1"})
            print(result.content[0].text)


anyio.run(main, "<token>")

streamable-http (/mcp) is the default and what current hosts expect. Set MCP_TRANSPORT=sse for older hosts (/sse); the guard is the same on both.

Local hosts over stdio

Claude Desktop, Cursor and Mistral Vibe can launch the server themselves. There is no request then, so the caller comes from the environment: the user's own access token, verified exactly like a bearer header would be, and bound to the process. Strict mode refuses to start without it. Queries run as TRINO_USER in service mode, and as the user, with that token, in jwt mode.

{
  "mcpServers": {
    "trino": {
      "command": "uvx",
      "args": ["akko-mcp-trino"],
      "env": {
        "MCP_TRANSPORT": "stdio",
        "TRINO_HOST": "trino.example.internal", "TRINO_PORT": "8443", "TRINO_HTTP_SCHEME": "https",
        "TRINO_USER": "mcp-reader", "TRINO_PASSWORD": "...",
        "MCP_AUTH_ENABLED": "true", "MCP_AUTH_REQUIRED": "true",
        "MCP_JWKS_URL": "https://idp.example.com/realms/data/protocol/openid-connect/certs",
        "MCP_OIDC_ISSUER": "https://idp.example.com/realms/data",
        "MCP_OIDC_AUDIENCE": "data-platform",
        "MCP_USER_TOKEN": "<the user's access token>"
      }
    }
  }
}

Use it from an agent

examples/agent.py is a complete agent in eighty lines: any OpenAI-compatible model (Mistral on La Plateforme, through OpenRouter or LiteLLM; or any other provider), the MCP tools, the user's token. The model never sees the token; the server receives it on every tool call and verifies it.

pip install akko-mcp-trino openai
export LLM_BASE_URL=https://api.mistral.ai/v1
export LLM_API_KEY=...
export LLM_MODEL=mistral-small-latest
export MCP_URL=http://localhost:3000/mcp
export USER_TOKEN=<the user's access token>
python examples/agent.py "Which catalogs can I see, and what is in them?"

The examples folder has the details.

The tools

Every tool runs in Trino through the identity mode, the listing tools included: metadata is data, and Trino lists only what that identity may see. Every description is written for a model and says what the tool returns. The descriptions of describe_table and execute_query also say that a masked value or a missing row is the access policy, not an error to retry.

Every identifier is validated ([A-Za-z_][A-Za-z0-9_-]*) before it is placed in SQL. execute_query accepts any SQL Trino accepts, as long as it is a read: the statement is parsed into an AST and refused if it is more than one statement, or if an INSERT, UPDATE, DELETE, MERGE, CREATE, DROP, ALTER, GRANT, CALL or SET appears anywhere in the tree, including inside a CTE, a subquery, or the statement an EXPLAIN explains. EXPLAIN ANALYZE is refused: it executes what it explains. Table functions and the system catalog are refused, and so is any SHOW that names system. A name without a catalog is read in the session catalog (TRINO_CATALOG, system by default); when that is system, a table without a catalog is refused (a CTE is not a table), and only SHOW CATALOGS and a SHOW SCHEMAS, SHOW TABLES or SHOW COLUMNS whose FROM gives the full name, catalog included, get through. So set TRINO_CATALOG to a data catalog; the engine's policy must refuse system too, the guard being a second line.

Results are capped at TRINO_MAX_ROWS: one row more is read, so truncated says whether the answer was cut, and the rest of the query is cancelled in Trino. Every listing says it too. With TRINO_QUERY_TIMEOUT_SECONDS, a query that runs longer is cancelled in Trino and the agent is told so. The budget is enforced by the server, not set as a session property: a policy engine that forbids users to set session properties would refuse every query otherwise. With MCP_CALL_TIMEOUT_SECONDS, a whole tool call has a time budget, every query it runs included: past it no query starts, the one running is cancelled (even when the budget ran out before Trino named it), and the agent gets {"error": "the tool call exceeded its time budget…"}, never what the call had gathered so far, nor an answer that came past the deadline. A plugin that calls its own backend bounds that request with akko_mcp_trino.audit.time_allowed(timeout): its own timeout, or what is left of the call's budget when that is shorter.

Errors come back to the agent as {"error": "..."} with Trino's whole message; a permission refusal from Trino is an ordinary error, not a crash. The audit line keeps only the error's code (see Audit).

Adding tools from another package

A product built on this server can add its own tools without forking it. A package registers a registrar (mcp, client) -> None, or (mcp, client, tools) -> None, under the entry-point group akko_mcp_trino.tools and the operator names it in MCP_TOOL_PLUGINS:

[project.entry-points."akko_mcp_trino.tools"]
my-tools = "my_package.tools:register"

from akko_mcp_trino.identity import current_subject


def register(mcp, client):
    @mcp.tool()
    def my_tool(text: str) -> str:
        return client.query("SELECT ...", user=current_subject())

Nothing is loaded that is not named, and an unknown name refuses to start. Tools added this way go through the same client, hence the same identity mode as the built-in ones, and every tool a registrar adds through mcp.tool() or mcp.add_tool() leaves the same audit line, with no change to the plugin. build_server(extra_tool_registrars=...) does the same from Python.

A registrar with a third argument receives the ToolContext: tools.tool(title, governed=..., open_world=...) registers a tool with its MCP annotations, tools.run(sql) queries Trino through the identity mode, tools.note(...) records what the call did (the SQL it ran, the tables, the rows, whether they were cut), and tools.add_describer(fn) adds to what describe_table answers:

import json


def register(mcp, client, tools):
    @tools.tool("Ask the data", governed=True, idempotent=False)
    def ask_data(question: str) -> str:
        answer = my_service.ask(question)  # runs its SQL through tools.run
        tools.note(sql=answer["sql"], tables=answer["tables"], rows=len(answer["rows"]))
        return json.dumps(answer)

Extensions

A tool plugin adds tools. An extension may add more, around the core and never in place of it. A package registers a factory (config, env) -> Extension under the entry-point group akko_mcp_trino.extensions, and the operator names it in MCP_EXTENSIONS (comma-separated, in order):

[project.entry-points."akko_mcp_trino.extensions"]
my-extension = "my_package.extension:create"

from akko_mcp_trino.extensions import Extension, Refusal


class MyExtension(Extension):
    name = "my-extension"

    def after_auth(self, admission):
        if admission.principal is not None and admission.principal.subject.startswith("svc-"):
            return Refusal("service_accounts_refused", status=403)
        return None


def create(config, env):
    return MyExtension()

Nothing is loaded that is not named. An unknown name refuses to start, and so does a factory that raises ValueError on a setting it cannot serve; --check reports both. akko_mcp_trino.extensions.Extension is the base class: every hook does nothing, so an extension overrides only what it adds.

A guard answers a Refusal(reason, status=401, error="", headers={}): the reason goes into X-Reason and the body, error is the RFC 6750 code a challenge may carry, headers are added to the response (Retry-After). It reads an Admission: the request headers, the token the request presented, the verified principal (None before authentication or without a token), stdio (true for the single admission of a stdio process); admission.bind(var, value) sets a context variable for the duration of the request, reset afterwards.

The rule: an extension can only refuse, never widen. Its guards run around the core's token verification, never instead of it; strict mode still refuses a request without a verified identity whatever a guard answers, and the read-only guard on SQL applies to every query. An identity mode decides who the server is for Trino, query by query (akko_mcp_trino.trino_client.resolve_identity picks the mode TRINO_IDENTITY_MODE names among the core's and the extensions', and refuses an unknown one: there is no fallback to service). A plugin that calls its own backend for the caller reads the verified bearer with akko_mcp_trino.identity.current_bearer(). Trino decides what that identity may read.

What happens on a request

sequenceDiagram
    participant H as MCP host
    participant G as Guard (ASGI middleware)
    participant T as Tool
    participant Tr as Trino

    H->>G: POST /mcp  Authorization, X-Request-Id
    G->>G: request id: honoured or generated
    opt an extension's guard before the token
        G-->>H: its refusal  X-Reason
    end
    G->>G: verify JWT (JWKS, iss, aud, exp) → Principal(subject)
    alt no verified identity and MCP_AUTH_REQUIRED
        G-->>H: 401  X-Reason: unauthenticated
    end
    opt an extension's guard after the token
        G-->>H: its refusal  X-Reason
    end
    G->>T: Principal in a ContextVar
    T->>T: validate identifiers · refuse writes
    T->>Tr: SQL as TRINO_USER (the identity mode), source
    Tr-->>T: rows (masked, filtered by the policy engine)
    T->>G: audit line {request_id, tool, subject, ok, error, duration_ms}
    G-->>H: result  X-Request-Id

Configuration

Everything is read from the environment. Nothing is hardcoded. An extension reads its own variables from the same environment.

Trino

Identity

Auth enabled without a JWKS URL refuses to start: an authentication layer that cannot verify anything must not pretend to.

Plugins and extensions

Serving

Operations

Refusals

Every refusal carries X-Reason and the request id, never the token. The core refuses with 401 unauthenticated a request without a verified identity in strict mode; an extension's guards add their own reasons.

Audit

Every response of the MCP endpoint carries X-Request-Id. It is the one the gateway in front of the server sent, when it is a plain token (letters, digits, ., _, :, -, at most 128), or one the server generated. A tool call outside an HTTP request (stdio) gets its own id. Every tool call, a plugin's included, writes one JSON line to the mcp.audit logger keyed by that id:

{"request_id":"edge-42","tool":"execute_query","subject":"alice","ok":false,"error":"USER_ERROR:PERMISSION_DENIED","duration_ms":84}

With the default logging configuration, the line comes out prefixed with INFO:mcp.audit:; a collector that expects pure JSON strips that prefix.

What a call did (the SHA-256 of its SQL, the tables, the rows, whether they were cut) is kept in its record while it runs (akko_mcp_trino.audit.note); the core's line does not write it, an extension may (wrap_audit).

Health and metrics

/health on the health port never touches Trino, so a liveness probe cannot kill the server because Trino is slow. /ready asks the identity mode: in service mode, a bounded SELECT 1 as TRINO_USER. /metrics exposes mcp_trino_queries_total, mcp_trino_query_errors_total and mcp_trino_query_duration_seconds.

Development

pip install -e ".[dev]"
ruff check akko_mcp_trino tests examples && ruff format akko_mcp_trino tests examples
ruff check akko_mcp_trino --select D100,D101,D102,D103,D105,D107   # every public name documented
mypy akko_mcp_trino           # the package ships py.typed and type-checks clean
pip-audit                     # no known vulnerability in the dependency tree
bash lint-vendor-neutral.sh   # fails if the package imports anything product-specific
pytest                        # 322 tests; line or branch coverage below 100 % fails the run
python -m build && twine check dist/*
akko-mcp-trino --check        # effective configuration, no secrets; exit 2 if it would not start

tests/test_core_alone.py lists what the core holds (tools, modules, settings, audit fields) and fails the day something is added unlisted.

Every module in akko_mcp_trino/ has one responsibility and its tests, in tests/:

The README is tested too: every variable listed here is read by config.py, and the defaults stated here are the defaults in the code.

Design notes

The guard is a pure ASGI middleware, not BaseHTTPMiddleware. The latter runs the downstream in a separate task and breaks ContextVar propagation: the verified caller would never reach the tools.

Mounting is transport-independent. One code path builds the transport app and mounts the guard, whatever the transport. The guard once lived only on the branch the default transport never took; a test now asserts it for both.

Every tool goes through the identity mode, not only execute_query. Listing catalogs, schemas and tables is reading metadata; a test walks every tool to check it reaches Trino the same way.

Read-only is decided on the AST, anywhere in the tree. A leading-keyword check lets WITH w AS (DELETE FROM t) SELECT 1 through. So did a root-node check, until a live adversarial proof caught it. The guard now walks the whole tree.

Table functions are refused. TABLE(pg.system.query(query => '...')) parses as a plain SELECT, yet it sends raw SQL to the source under the connector's own account, so the engine applies neither row filters nor column masks. The guard refuses every table function and the system catalog (which shows other users' SQL in system.runtime.queries). The engine policy must refuse them as well; the guard is a second line, not the only one.

The verified identity, never one the caller names. In service mode every query runs as TRINO_USER, and the caller's token never reaches Trino. In jwt mode Trino receives the caller's token only once the server has verified it, and verifies it again. Either way, a caller sending a header that names a Trino user changes nothing. A password or a token is never sent over plain http, and an identity mode the core does not know refuses to start rather than fall back.

An extension can only refuse. Its guards run around the token verification, never instead of it, and the read-only guard applies whatever they answer.

The server does not decide access. Trino does, for the configured identity or for each caller. Putting policy in the server would duplicate, and eventually contradict, what the engine already enforces for every other client.

Contributing

Contributions are welcome: see CONTRIBUTING.md for the ground rules (tests first, 100 % coverage, product-neutral, fail closed) and SECURITY.md for reporting a vulnerability privately. Changes are tracked in CHANGELOG.md.

About AKKO

akko-mcp-trino is built and maintained by AKKO, a French project working on governed access for AI agents to enterprise data, in place, on the engines and identity providers organisations already run.

Maintained by: AKKO (contact@akko-ai.com) · Issues: github.com/AKKO-p/akko-mcp-trino/issues

License

Apache 2.0. See LICENSE.