Examples

Connect an MCP host, write an agent, check the server, all while keeping the user's identity.

These examples apply to the open source core in jwt mode as well as to the server delivered in AKKO and ARCHE. They all follow the same rule: the model never sees the person’s identity; the server receives it, Trino decides.

1. An MCP host (Cursor, Claude Desktop, VS Code)

The server runs on your network, behind HTTPS. The host connects to its address and presents the user’s identity on every request, here as an Authorization header.

{
  "mcpServers": {
    "trino": {
      "url": "https://mcp.example.com/mcp",
      "headers": {"Authorization": "Bearer <the user's access token>"}
    }
  }
}

2. An agent on an OpenAI-compatible model (Mistral, or any other)

A complete agent fits in a few dozen lines. It lists the server’s tools, exposes them to the model as function calls, runs each call over MCP and returns the answer. It works with Mistral on La Plateforme, through OpenRouter or LiteLLM, and with any provider that speaks the chat completion API with tools.

The agent’s system prompt tells the model two things, and they matter: explore the catalog before writing SQL, and take a masked value for what it is, the access policy, without retrying the query. Without the second, a model that receives ***@example.com may think it is an error and loop; we saw it happen.

Run twice for two people, the agent gives two different answers to the same question.

3. From Python, with the official SDK

An agent that chains its own steps lists what it sees, then runs a query. Every answer is a JSON object.

import json

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}"})
    async with streamable_http_client("https://mcp.example.com/mcp", http_client=http) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            async def call(tool, **args):
                result = await session.call_tool(tool, args)
                return json.loads(result.content[0].text)

            catalogs = await call("list_catalogs")
            # {"catalogs": [...], "truncated": false}
            rows = await call("execute_query", sql="SELECT country, count(*) FROM core_postgres.clients.customers GROUP BY country", max_rows=20)
            # {"columns": [...], "rows": [...], "row_count": n, "truncated": false}
            print(catalogs, rows)


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

When truncated is true, the agent should narrow its request (one catalog, a filter, an aggregation) rather than trust the list as complete.

4. Checking the server

# alive, reaches Trino, exposes metrics
curl -s localhost:3001/health ; curl -s localhost:3001/ready ; curl -s localhost:3001/metrics | head

/health never touches Trino, so that a liveness probe cannot kill the server because Trino is slow.

What is proven

The proof scripts run against a real cluster and can be replayed: two accounts, the same tools, two answers. The akko-mcp-trino page sums up what they check.