Exemples

Brancher un MCP host, é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, le mode par défaut, 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 MCP host par URL (Cursor)

Le serveur tourne sur votre réseau, derrière HTTPS. Le MCP host 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. Le token doit être émis pour ce serveur : un token émis pour Trino est refusé.

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

L’en-tête écrit à la main est un raccourci de test. En production, posez MCP_RESOURCE_URL : un MCP host qui implémente l’autorisation MCP obtient le token lui-même. VS Code a son propre format de fichier (clé servers). Le fichier de Claude Desktop ne lance que des serveurs locaux : il rejoint ce serveur par un pont local, comme le montre le guide Brancher Claude Desktop ou Cursor.

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

Un agent complet tient en quelques dizaines de lignes. Il liste les tools du serveur, les expose au modèle au format des function calls, 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 de modèle qui parle l’API de complétion de conversation avec tools.

Le prompt système de l’agent dit deux choses au modèle, et elles comptent : trouver où vit la donnée (search_columns, puis describe_schema) avant d’écrire du SQL, et prendre une valeur masquée pour ce qu’elle est, la policy 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 trouve où vit la donnée, apprend le schéma, exécute une requête, et parcourt les catalogs quand il veut voir ce qu’il y a. 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)

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

Quand une liste ou describe_schema rend un next_offset, rappelez le tool avec offset=next_offset pour lire la page suivante ; quand search_columns rend un next_cursor, rappelez-le avec cursor=next_cursor. Quand execute_query rend truncated: true, resserrez la requête (un filtre, une agrégation) plutôt que de croire les lignes complètes.

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 tools, deux réponses. La page akko-mcp-trino résume ce qu’ils vérifient.