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, the default, 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 over a URL (Cursor)

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. The token must be issued for this server: a token issued for Trino is refused.

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

A header written by hand is a testing shortcut. In production, set MCP_RESOURCE_URL: an MCP host that implements MCP authorization gets the token itself. VS Code has its own file format (the servers key). Claude Desktop’s file starts local servers only: it reaches this server through a local bridge, as the guide Connect Claude Desktop or Cursor shows.

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: find where the data lives (search_columns, then describe_schema) 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 finds where the data lives, learns the schema, runs a query, and browses the catalogs when it wants to see what there is. 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)

            found = await call("search_columns", text="customer country")
            # {"tables": [...], "matches": [{"catalog": "core_postgres", "schema": "clients", "table": "customers", "column": "country", ...}],
            #  "schemas_searched": n, "not_searched": [], "complete": true, "next_cursor": null}
            schema = await call("describe_schema", catalog="core_postgres", schema="clients")
            # {"catalog": "core_postgres", "schema": "clients", "tables": [{"table", "type", "comment", "columns": [...]}],
            #  "next_offset": null, "shortened": 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}
            catalogs = await call("list_catalogs")
            # {"catalogs": [...], "next_offset": null, "truncated": false}
            print(found, schema, rows, catalogs)


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

When a listing or describe_schema answers a next_offset, call again with offset=next_offset to read the next page; when search_columns answers a next_cursor, call again with cursor=next_cursor. When execute_query answers truncated: true, narrow the query (a filter, an aggregation) rather than trust the rows 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.