Exemples

Brancher un hôte MCP, écrire un agent, vérifier le serveur, tout en gardant l'identité de l'utilisateur.

Ces exemples valent pour le cœur open source en mode jwt comme pour le serveur livré dans AKKO et ARCHE. Ils suivent tous la même règle : le modèle ne voit jamais l’identité de la personne ; le serveur la reçoit, Trino décide.

1. Un hôte MCP (Cursor, Claude Desktop, VS Code)

Le serveur tourne sur votre réseau, derrière HTTPS. L’hôte se connecte à son adresse et présente l’identité de l’utilisateur à chaque requête, ici sous la forme d’un en-tête Authorization.

{
  "mcpServers": {
    "trino": {
      "url": "https://mcp.example.com/mcp",
      "headers": {"Authorization": "Bearer <le jeton d'accès de l'utilisateur>"}
    }
  }
}

2. Un agent sur un modèle compatible OpenAI (Mistral, ou autre)

Un agent complet tient en quelques dizaines de lignes. Il liste les outils du serveur, les expose au modèle au format des appels de fonctions, exécute chaque appel par MCP et rend la réponse. Il marche avec Mistral sur La Plateforme, via OpenRouter ou LiteLLM, et avec tout fournisseur qui parle l’API de complétion de conversation avec outils.

Le prompt système de l’agent dit deux choses au modèle, et elles comptent : explorer le catalogue avant d’écrire du SQL, et prendre une valeur masquée pour ce qu’elle est, la politique d’accès, sans relancer la requête. Sans la seconde, un modèle qui reçoit ***@example.com peut croire à une erreur et boucler ; nous l’avons vu.

Lancé deux fois pour deux personnes, l’agent rend deux réponses différentes à la même question.

3. Depuis Python, avec le SDK officiel

Un agent qui enchaîne ses propres étapes liste ce qu’il voit, puis exécute une requête. Chaque réponse est un objet JSON.

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, "<jeton>")

Quand truncated vaut true, l’agent doit resserrer sa demande (un catalogue, un filtre, une agrégation) plutôt que de croire la liste complète.

4. Vérifier le serveur

# le serveur est vivant, joint Trino, expose ses métriques
curl -s localhost:3001/health ; curl -s localhost:3001/ready ; curl -s localhost:3001/metrics | head

/health ne touche jamais Trino, pour qu’une sonde de vie ne tue pas le serveur parce que Trino est lent.

Ce qui est prouvé

Les scripts de preuve tournent contre un cluster réel et sont rejouables : deux comptes, les mêmes outils, deux réponses. La page akko-mcp-trino résume ce qu’ils vérifient.