AKKO MCP server for Trino

akko-mcp-trino

English · Français

让 AI Agent 只读访问 Trino，仅此而已。

akko-mcp-trino 是一个面向 Trino 的 MCP 服务。 它允许 Agent 列出目录、架构和表，查看表结构，解释并执行查询；对于收到的任何写入操作，它都会拒绝：每条语句都会经过解析，并在发送给 Trino 之前拒绝其中的任何写入。默认情况下，每条查询都以 Agent 所代表用户的身份运行。服务会验证该用户为此服务签发的令牌，并从身份提供方获取单独的 Trino 令牌（OAuth 2.0 令牌交换）：用户自己的令牌绝不会离开服务。随后，Trino 的访问控制（OPA、Ranger 或 Trino 自身规则）会决定该用户可以读取哪些内容。结果在行数和执行时间上均有限制，每次调用都会留下一条审计日志。

从 0.5 升级？ 0.6.0 更改了默认设置：必须验证调用方身份，并且 jwt 模式会交换调用方的令牌，而不是直接传递该令牌。请参阅 MIGRATING.md。

原因

需要读取湖仓的 Agent 需要一个小巧、可预测的接口：少量模型能够理解其描述的工具，不会无限膨胀的答复，以及一项保障，确保提示词注入不会借此把一个问题变成 DROP TABLE。这个服务提供的正是这样的接口，仅此而已。它不负责决定访问权限：Trino 会根据已有规则为每位用户做出决定；或者在 service 模式下，根据你提供的唯一身份做出决定。

亮点

• 通过此服务，设计上即为只读。 SQL 会被解析为 AST；如果树中的任何位置出现写操作，都会被拒绝，包括 CTE 和子查询。表函数（TABLE(...)，例如 JDBC 连接器的 system.query）和 system 目录也会被拒绝；当会话目录为 system 时，即使使用不带目录的名称也会被拒绝。用户通过其他方式在 Trino 中可以执行哪些操作，仍由 Trino 的规则决定：请参阅只读保护的范围。
• 默认安全，遵循用户自身的权限。 在默认的 jwt 模式下，服务会验证调用者的令牌（为此服务签发，绝不会为 Trino 签发），然后在身份提供方处将其交换为受众为 Trino 的令牌（RFC 8693），由 Trino 以该用户身份执行查询。调用者的令牌绝不会被传递。没有有效令牌的请求会被拒绝，并返回 401。
• 无需令牌交换的备用方案。 在 impersonation 模式下，服务使用自己的账户（密码或客户端证书）向 Trino 进行身份验证，Trino 则依据仅该账户拥有的模拟规则，以经过验证的调用者身份执行查询。
• 只有明确指定时才使用单一身份。 在 service 模式下，每个查询都以 TRINO_USER 身份运行。该模式绝不会作为默认选项；服务会在启动时以及向每个主机发送的初始化说明中明确说明这一点。
• 默认设有限制。 行数上限为 100，并附带 truncated 标记；每个查询的时间预算为 60 s，超时后会在 Trino 中取消；每次工具调用的时间预算为 120 s。
• 每次调用记录一条审计日志。 以 X-Request-Id 为键，绝不记录令牌或 SQL 文本：记录谁调用了哪个工具、结果如何，以及耗时多久。
• 仅有六个工具， 并明确标注其实际能力：只有在 TRINO_READ_ONLY 为 true 时，execute_query 才会标注为只读。
• 设有防护的入口。 Bearer 令牌会依据任意 OIDC 提供方的 JWKS 进行验证（签发者、绑定到此服务的受众、过期时间、时钟容差）；为 Trino 签发的令牌会被拒绝；提供受保护资源元数据（RFC 9728）和 Host 标头过滤器。
• 可扩展，无需派生分支。 工具插件和扩展是启动时指定的 Python 包，可用于配置身份模式、入口防护、审计字段、路由和工具。扩展只能拒绝核心允许的操作，绝不能扩大权限范围。
• 适用于任何 MCP 主机和任何模型。 Cursor、Claude Desktop、VS Code、Python SDK……Mistral，或通过示例 Agent 使用任何兼容 OpenAI 的模型。
• 精简且经过验证。 共 377 项测试，行覆盖率和分支覆盖率均为 100 %；包含针对真实 SDK 的进程内测试套件，覆盖全部三种传输方式；在 CI 中针对 Keycloak 和 Trino 进行实际运行（integration/），并包含与产品无关的防护机制。

文档

• 端到端了解其工作原理： akko-ai.com/en/docs/how-it-works（另有法语版）。
• 指南、前置条件、配置和示例： akko-ai.com/docs/akko-mcp-trino（另有英文版）以及 akko-ai.com/docs/exemples。
• 本仓库中的参考资料： 本 README、examples/、CHANGELOG.md、MIGRATING.md、CONTRIBUTING.md、SECURITY.md。

目录

1. 工作原理
2. 兼容性
3. 前置条件
4. 安装和运行
5. 连接 MCP 主机
6. 从 Agent 使用
7. 工具
8. 只读保护的范围
9. 扩展
10. 请求处理流程
11. 配置
12. 运维
13. 开发
14. 设计说明
15. 贡献
16. 关于 AKKO

工作原理

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

    Host -- "1. tool call + Bearer token<br/>(audience: this server)" --> Server
    Server -. "2. verify signature, issuer,<br/>audience, expiry (JWKS)" .-> IdP
    Server -. "3. exchange it for a token<br/>whose audience is Trino (RFC 8693)" .-> IdP
    Server -- "4. read-only SQL, with that token" --> Trino
    Trino -- "5. may this user read this?" --> Policy
    Trino -- "6. rows, as that user may read them" --> Server
    Server -- "7. result" --> Host
    Trino --> Data

宿主使用用户的访问令牌调用工具。服务会验证令牌（根据提供方的 JWKS 验证签名、颁发者、指向此服务的受众以及有效期），若请求没有有效令牌，则返回 401 并拒绝请求；针对 Trino 颁发的令牌也会被拒绝。随后，服务会检查语句（只读、标识符、表函数、system 目录），并以 TRINO_IDENTITY_MODE 指定的身份模式在 Trino 中执行语句。Trino 会向其策略引擎查询，并返回该身份获准读取的内容，同时限制行数和执行时间。

模式	Trino 看到的身份	Trino 如何识别	适用场景
jwt（默认）	调用者	服务通过令牌交换（RFC 8693）获取一个面向 Trino 的令牌；Trino 使用其 JWT 认证器验证该令牌	身份提供方支持 OAuth 2.0 Token Exchange（RFC 8693）；已使用 Keycloak 26 测试
impersonation	调用者	服务自己的账户（密码或客户端证书）请求 Trino 以调用者身份运行查询；Trino 仅在该账户专属的模拟规则允许时才会执行	身份提供方无法交换令牌
service	TRINO_USER，所有调用者共用同一身份	在本地使用密码、客户端证书或不使用凭据	仅在明确指定时使用：所有人共用一个身份，启动时会声明该身份，并告知每个宿主

无论使用哪种模式，调用者自己的令牌都不会离开服务：MCP 授权规范禁止转发该令牌（“令牌直通”）。调用者的名称始终来自经过验证的令牌，而不是其自行编写的标头。以调用者身份运行的两种模式都要求入口必须强制验证有效令牌，否则服务不会启动。扩展（见扩展）可以添加其他身份模式。

兼容性

	支持	已测试
Python	3.12、3.13	3.12、3.13（CI）
Trino	351 及更高版本；jwt 模式使用 https 和 JWT 认证器，impersonation 模式使用密码或证书认证器及模拟规则；service 模式使用 http（本地）或 https	最近的版本，使用 JWT 和证书认证器，在 OPA 和基于文件的规则后运行，并启用了行筛选和列屏蔽
策略引擎	OPA（trino-opa）、Ranger（Trino 插件）、Trino 基于文件的访问控制	OPA 和基于文件的规则，启用了行筛选和列屏蔽
身份提供方	发布 JWKS 的任何 OIDC 提供方（入口）；OAuth 2.0 Token Exchange，RFC 8693（jwt 模式）	Keycloak 26（标准令牌交换）
MCP	由官方 mcp SDK 2.2 协商的协议（最高至 2025-11-25 版本）；传输方式为 streamable-http、sse 和 stdio	三种传输方式均已测试，使用官方 SDK 客户端并在 CI 中运行
MCP 宿主	远程：任何发送 bearer 标头的宿主；本地：任何 stdio 宿主	Python SDK、示例 Agent
模型	任意；服务本身不会与模型通信	通过 OpenRouter 使用 Mistral Small 3.2 驱动这些工具

前提条件

你需要	用途	说明
Python 3.12 或更高版本	运行时	pip 和虚拟环境
可访问的 Trino coordinator	引擎	任意近期版本；使用 jwt 和 impersonation 时，以及发送任何密钥时，都需要 https
带有 JWKS 端点的 OIDC 提供商	入口	Keycloak、Entra ID、Okta、Dex……令牌必须包含 iss、一个指向此服务的 aud、exp，以及一个主体（preferred_username 或 sub）
jwt：在提供商处进行令牌交换，以及 Trino 的 JWT 身份验证器	每位用户以自己的权限读取	为此服务创建一个机密客户端，并允许它将为该服务签发的令牌交换为受众为 Trino 的令牌；设置 http-server.authentication.type=JWT，并配置提供商的 JWKS 和 required-audience，使其指向该受众
或 impersonation：此服务的账户，以及一条模拟规则	每位用户以自己的权限读取，无需令牌交换	TRINO_USER 和 TRINO_PASSWORD，或客户端证书；配置一条 Trino 规则，允许且仅允许该账户模拟用户
或 service：一个 Trino 身份	每位调用者都使用该身份的权限	TRINO_USER，以及 TRINO_PASSWORD 或客户端证书；仅授予它对所有调用者都可读取内容的读取权限
为 Trino 决策的策略引擎	治理	OPA（opa.policy.uri）、Ranger 或 Trino 基于文件的规则
一个 MCP 主机	客户端	Cursor、Claude Desktop、VS Code，或任何基于 mcp SDK 构建的客户端

可选：用于抓取 /metrics 的 Prometheus。

安装并运行

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

或者从源代码安装：

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 .

此软件包会安装一个 Python 模块（akko_mcp_trino）和一个命令 （akko-mcp-trino）；python -m akko_mcp_trino 是相同的入口点。

配置 Trino 和身份提供商，然后启动服务。默认设置是安全的：每个查询都以调用者的身份运行（jwt），没有有效令牌的请求会被拒绝。

export TRINO_HOST=trino.example.internal
export TRINO_PORT=8443
export TRINO_HTTP_SCHEME=https          # required: a token never travels over http
export TRINO_CATALOG=iceberg

# The door: tokens issued for this server, verified against the provider's JWKS
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=akko-mcp        # the audience the provider gives tokens for this server
export MCP_RESOURCE_URL=https://mcp.example.com/mcp   # optional: this server's canonical URL

# A token for Trino, by token exchange (RFC 8693), never the caller's own
export TRINO_TOKEN_EXCHANGE_URL=https://idp.example.com/realms/data/protocol/openid-connect/token
export TRINO_TOKEN_EXCHANGE_CLIENT_ID=akko-mcp
export TRINO_TOKEN_EXCHANGE_CLIENT_SECRET=...
export TRINO_TOKEN_EXCHANGE_AUDIENCE=trino

python -m akko_mcp_trino

启动服务前，akko-mcp-trino --check 会打印生效的配置（绝不包含密钥）和任何警告；如果存在会导致服务无法启动的问题，则以 2 退出；使用 stdio 时，它会检查是否存在 MCP_USER_TOKEN，而不会验证其有效性（真正启动时会验证令牌）。然后：

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

检查服务是否正常运行且能够访问 Trino：

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

每位用户以自己的权限读取（jwt，默认模式）

调用者的令牌是为此服务签发的：其受众为 MCP_OIDC_AUDIENCE（或 MCP_RESOURCE_URL）。服务会在提供商处将该令牌交换为受众为 TRINO_TOKEN_EXCHANGE_AUDIENCE 的令牌，在令牌即将过期前将其保留在内存中，并发送给 Trino。受众为 Trino 的令牌会在入口处被拒绝，任何会接受此类令牌的配置都会导致服务无法启动。

在提供商处，为此服务创建一个允许进行令牌交换的机密客户端，并向其提供 Trino 的受众。使用 Keycloak 26 时：为服务的客户端启用 Standard token exchange，添加指向 Trino 客户端的受众映射器，并让 MCP 主机的客户端将服务的客户端加入其所获取令牌的受众中（同样使用受众映射器）。integration/ 文件夹包含了一个此类 realm，并且每个拉取请求都会对其进行测试。

在 Trino 端，设置 http-server.authentication.type=JWT， 将 http-server.authentication.jwt.key-file 设置为提供商的 JWKS URL， 将 http-server.authentication.jwt.principal-field 设置为与服务读取的相同声明（preferred_username 或 sub），并通过 required-issuer 和 required-audience 将令牌限定为来自你的提供商且受众为 Trino 的令牌。两个用户提出相同问题时，会分别获得其自身权限所允许的答案。由于服务本身没有可用于查询的身份，/ready 会请求 Trino 的公共 /v1/info。

无令牌交换（impersonation）

如果提供方无法交换令牌，服务会以自身账户连接 Trino，而 Trino 会以经过验证的调用者身份运行每个查询：

export TRINO_IDENTITY_MODE=impersonation
export TRINO_USER=mcp-service
export TRINO_CLIENT_CERT=/certs/mcp-service.crt   # mTLS; or TRINO_PASSWORD over https
export TRINO_CLIENT_KEY=/certs/mcp-service.key

Trino 必须对该账户进行身份验证（除用户使用的身份验证器外，还需配置证书或密码身份验证器），并且只允许该账户模拟用户。使用基于文件的访问控制时：

{"impersonation": [{"original_user": "mcp-service", "new_user": ".*", "allow": true}]}

将 new_user 限定为 Agent 可以代为操作的用户。其他人一律不得模拟用户：如果用户发送的标头中指定了其他人，Trino 会拒绝该请求；服务不会从标头中获取调用者姓名，只会使用其经过验证的令牌。请像保管其他服务凭据一样保管该账户的凭据：任何持有它的人都能以规则允许的任何用户身份行事。

所有调用者使用同一身份（仅在明确指定时使用 service）

export TRINO_IDENTITY_MODE=service
export TRINO_USER=mcp-reader
export TRINO_PASSWORD=...               # or TRINO_CLIENT_CERT and TRINO_CLIENT_KEY; https only

无论调用者是谁，每个查询都会以 TRINO_USER 身份运行：为该身份授予本服务所有调用者可能需要的权限，但不得授予更多权限，并且只能授予只读权限。服务启动时会记录一条警告，--check 会显示该警告，每个主机都会在 MCP 初始化的 instructions 中收到该警告。默认情况下，身份验证门仍会启用；如需在本地禁用它，同时设置 MCP_AUTH_ENABLED=false 和 MCP_AUTH_REQUIRED=false，警告中也会说明这一点。

容器

仓库随附一个 Dockerfile（非 root 用户运行、包含健康检查和 OCI 标签）。使用相同的环境变量运行，并发布端口 3000（MCP）和 3001（健康检查）。

连接 MCP 主机

主机会发送用户从身份提供方获取、专用于本服务的访问令牌；服务会验证该令牌，并在审计日志中记录调用者身份。支持 MCP 授权的主机会通过服务在设置 MCP_RESOURCE_URL 时发布的受保护资源元数据发现身份提供方。

Cursor、Claude Desktop、VS Code（mcp.json）：

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

使用官方 SDK 从 Python 调用：

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），这是当前主机所需的传输方式。对于较旧的主机，设置 MCP_TRANSPORT=sse（/sse）；两种传输方式使用相同的身份验证门。

通过 stdio 连接本地主机

Claude Desktop、Cursor 和 Mistral Vibe 可以自行启动服务。此时没有请求，因此调用者身份来自环境：用户自己的访问令牌，专用于本服务，其验证方式与 bearer 标头完全相同，并绑定到进程。默认的严格模式会在没有该令牌时拒绝启动。在 jwt 模式下（令牌会交换为 Trino 令牌，与通过 http 时相同）和 impersonation 模式下，查询会以用户身份运行；在 service 模式下，则以 TRINO_USER 身份运行。

{
  "mcpServers": {
    "trino": {
      "command": "uvx",
      "args": ["akko-mcp-trino"],
      "env": {
        "MCP_TRANSPORT": "stdio",
        "TRINO_HOST": "trino.example.internal", "TRINO_PORT": "8443", "TRINO_HTTP_SCHEME": "https",
        "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": "akko-mcp",
        "TRINO_TOKEN_EXCHANGE_URL": "https://idp.example.com/realms/data/protocol/openid-connect/token",
        "TRINO_TOKEN_EXCHANGE_CLIENT_ID": "akko-mcp",
        "TRINO_TOKEN_EXCHANGE_CLIENT_SECRET": "...",
        "TRINO_TOKEN_EXCHANGE_AUDIENCE": "trino",
        "MCP_USER_TOKEN": "<the user's access token>"
      }
    }
  }
}

在 Agent 中使用

examples/agent.py 是一个完整的 Agent，只有 80 行代码：可使用任何兼容 OpenAI 的模型（La Plateforme 上的 Mistral、通过 OpenRouter 或 LiteLLM 使用，或使用其他提供方）、MCP 工具和用户令牌。模型永远不会看到令牌；服务会在每次工具调用时接收令牌并进行验证。

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

详情请参阅 examples 文件夹。

工具

工具	参数	返回值
list_catalogs	无	{"catalogs": [...], "truncated": bool}：Trino 显示的目录
list_schemas	catalog	{"schemas": [...], "truncated": bool}：该目录中的架构
list_tables	catalog、schema	{"tables": [...], "truncated": bool}：该架构中的表和视图
describe_table	catalog、schema、table、sample_rows（0 到 20）	列、类型、注释、truncated；按需提供受管控的样本；以及扩展的描述器添加的内容
explain_query	sql	{"plan": "..."}：Trino 将如何运行只读语句，但不会实际运行
execute_query	sql、max_rows（可选，且必须低于服务器上限）	{"columns": [...], "rows": [...], "row_count": n, "truncated": bool}
所有工具都以身份模式通过 Trino 运行，包括列表工具：元数据也是数据，Trino 只会列出该身份可见的内容。每条描述都面向模型编写，并说明工具会返回什么。describe_table 和 execute_query 的描述还会说明，值被掩码或行缺失是访问策略的结果，而不是需要重试的错误。

每个标识符都会先通过验证（[A-Za-z_][A-Za-z0-9_-]*），再放入 SQL。execute_query 接受 Trino 支持的任意 SQL，前提是只读：语句会被解析为 AST，如果包含多条语句，或树中任何位置出现 INSERT、UPDATE、DELETE、MERGE、CREATE、DROP、ALTER、GRANT、CALL 或 SET，就会被拒绝，包括这些语句出现在 CTE、子查询或 EXPLAIN 所解释的语句中。EXPLAIN ANALYZE 会被拒绝，因为它会执行所解释的语句。表函数和 system 目录会被拒绝，任何指定了 system 的 SHOW 语句也会被拒绝。未指定目录的名称会在会话目录中读取（默认为 TRINO_CATALOG，即 system）；当会话目录为 system 时，未指定目录的表会被拒绝（CTE 不是表），只有 SHOW CATALOGS，以及 FROM 指定完整名称（包括目录）的 SHOW SCHEMAS、SHOW TABLES 或 SHOW COLUMNS 才能通过。因此，请将 TRINO_CATALOG 设为数据目录；引擎策略也必须拒绝 system，这道防护只是第二道防线。

结果最多返回 TRINO_MAX_ROWS 行（默认值为 100，最少为 1）：系统会多读取一行，因此 truncated 会指出答案是否被截断，其余查询会在 Trino 中取消。每个列表工具也会说明这一点。通过 TRINO_QUERY_TIMEOUT_SECONDS（默认值为 60 s），运行时间更长的查询会在 Trino 中取消，并告知 Agent。此预算由服务端强制执行，而非设为会话属性：否则，禁止用户设置会话属性的策略引擎会拒绝所有查询。通过 MCP_CALL_TIMEOUT_SECONDS（默认值为 120 s），整个工具调用都有时间预算，包括它运行的每个查询：超时后不会启动新查询，正在运行的查询会被取消（即使预算在 Trino 为其命名之前就已耗尽），并且 Agent 会收到 {"error": "the tool call exceeded its time budget…"}，而不会收到调用迄今收集到的内容，也不会收到截止时间之后才得出的答案。调用自有后端的插件会使用 akko_mcp_trino.audit.time_allowed(timeout) 限制该请求的时长：采用插件自身的超时，或在调用预算剩余时间更短时采用剩余时间。

这些默认值是为 Agent 选择的，而非为批处理作业选择的。Agent 提出的问题应该在一分钟内得到答复；需要更长时间的查询通常是在进行全表扫描，Agent 应该缩小查询范围，收到的错误也会说明这一点。每次调用两分钟的预算为运行两三个查询的工具留出了余地，同时避免 Agent 的一轮交互卡在单个工具上。将任一项设为 0 即可移除限制（此时查询会运行到 Trino 自身的 query.max-execution-time 允许的时长）；--check 会打印两项设置。

只有在 TRINO_READ_ONLY 为 true 时，execute_query 才会标注 readOnlyHint: true。当 TRINO_READ_ONLY=false 时，其注解会说明它可能写入（readOnlyHint: false、destructiveHint: true），描述也会作出相同说明，以便宿主请求确认，而不是像对只读工具那样跳过确认。无论该设置为何值，其他五个工具都只会读取。

错误会以 {"error": "..."} 的形式返回给 Agent，并包含 Trino 的完整消息；Trino 拒绝权限属于普通错误，不会导致崩溃。审计日志行只保留错误代码（参见审计）。

从其他软件包添加工具

基于此服务构建的产品可以添加自己的工具，无需分叉该项目。软件包可在入口点组 akko_mcp_trino.tools 下注册一个 (mcp, client) -> None 或 (mcp, client, tools) -> None 注册器，运维人员则在 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())

未指定名称的工具不会加载，遇到未知名称时服务将拒绝启动。通过这种方式添加的工具会使用同一个客户端，因此身份模式也与内置工具相同；注册器通过 mcp.tool() 或 mcp.add_tool() 添加的每个工具都会留下相同的审计记录，无需更改插件。 build_server(extra_tool_registrars=...) 在 Python 中也能实现相同效果。

带有第三个参数的注册器会收到 ToolContext： tools.tool(title, governed=..., open_world=...) 会注册带有 MCP 注解的工具，tools.run(sql) 会通过该身份模式查询 Trino，tools.note(...) 会记录调用执行的操作（运行的 SQL、涉及的表、行数以及结果是否被截断），而 tools.add_describer(fn) 会扩展 describe_table 的回答：

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)

只读保护的范围

只读保护涵盖经由此服务执行的操作：如果 Agent 发送的语句中任何位置出现写操作，该语句都会被拒绝，不会发送给 Trino。这并不保证用户无法通过其他方式在 Trino 中执行操作。

在 jwt 和 impersonation 模式下，Trino 会以用户身份运行每个查询，并遵循策略授予该用户的权限。如果策略允许用户写入，用户仍可通过其他 Trino 客户端（CLI、笔记本、BI 工具）写入，只需使用服务提供方直接提供给他们的 Trino 令牌。服务通过交换获得的令牌会保存在内存中，绝不会返回给 Agent 或主机，但用户在 Trino 中拥有的权限不会因此改变。在 service 模式下，TRINO_USER 也是如此：持有其凭据的人也拥有相应权限。

若要从头到尾实现只读，也要确保引擎本身贯彻这一限制：

• service：仅授予 TRINO_USER 读取权限。使用基于文件的访问控制时，配置为 {"catalogs": [{"user": "mcp-reader", "catalog": ".*", "allow": "read-only"}]}； 使用 OPA 或 Ranger 时，则创建一条仅授予该身份 SELECT 权限的策略。
• impersonation：将模拟规则限制为 Agent 可代为操作的用户；如果这些用户可以在其他地方写入，则确保服务账户自身除 SELECT 1 外不拥有任何权限。
• jwt：仅允许此服务的客户端通过交换获取面向 Trino 受众的令牌，这样 Agent 所在主机持有的令牌就无法访问 Trino；并在策略引擎中，为 Agent 可访问的目录配置只读规则，适用于通过 Agent 触达的用户。无论查询由哪个客户端发送，Trino 都会应用相同的规则，因此应在策略中确保用户通过所有路径（包括此路径）都只能读取。

保护措施仍是第二道防线，而非唯一防线：引擎策略也必须拒绝表函数和 system 目录（请参阅设计说明）。

扩展

工具插件会添加工具。扩展可以围绕核心功能添加更多内容，但绝不能取代核心功能。软件包会在入口点组 akko_mcp_trino.extensions 下注册一个工厂 (config, env) -> Extension，运维人员再通过 MCP_EXTENSIONS（以逗号分隔，按顺序排列）指定它：

[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()

未指定的扩展不会加载。遇到未知名称时，服务会拒绝启动；工厂遇到无法处理的设置而引发 ValueError 时，也会拒绝启动；--check 会报告这两种情况。akko_mcp_trino.extensions.Extension 是基类：所有钩子默认都不执行任何操作，因此扩展只需重写要添加功能的钩子。

钩子	添加的内容
check_lines()	--check 输出的行，不包含任何机密
identity_modes()	由 TRINO_IDENTITY_MODE 名称指定的身份模式，每种模式都是一个 IdentityMode（describe、connection(user)、ping）；若名称与某个核心模式（jwt、impersonation、service）相同，则替换该模式
audiences()	除 MCP_OIDC_AUDIENCE 和 MCP_RESOURCE_URL 之外接受的令牌受众；不包括 Trino 的受众
before_auth(admission)	在检查令牌之前拒绝请求，或返回 None
after_auth(admission)	核心验证令牌（或确认没有令牌）后拒绝请求，或返回 None
public_documents()	按请求路径提供无需令牌即可访问的 JSON 文档；优先于核心文档
challenge(refusal)	拒绝响应的 WWW-Authenticate 值；优先于核心值
routes()	添加到服务应用中的 Starlette 路由，受身份验证保护
wrap_audit(audit)	服务端写入的审计接收器，以原本要使用的接收器为输入
client_tags()	当前查询所用的客户端标签（X-Trino-Client-Tags）
describe_table_doc()	附加到 describe_table 描述中的文本
tool_registrars()	工具注册器，调用方式与工具插件的注册器相同，并先于插件调用

守卫会返回 Refusal(reason, status=401, error="", headers={}, message="")：reason 会写入 X-Reason 和响应正文；error 是挑战响应可携带的 RFC 6750 错误代码；headers 会添加到响应中（例如 Retry-After）；如果设置了 message，它会在响应正文中告知客户端应如何处理。守卫读取一个 Admission：请求头、请求提交的令牌、已验证的 principal（身份验证前或请求未携带令牌时为 None）、stdio（stdio 进程的唯一一次准入时为 true）；admission.bind(var, value) 会在请求期间设置上下文变量，并在之后重置。

规则：扩展只能拒绝请求，绝不能扩大权限。 它们的守卫会围绕核心令牌验证运行，绝不会取代核心验证；严格模式下，无论守卫返回什么，没有已验证身份的请求仍会被拒绝；SQL 的只读守卫适用于每条查询。身份模式决定服务端向 Trino 使用的身份，并按查询逐次决定（akko_mcp_trino.trino_client.resolve_identity 会从核心和扩展的身份模式中选择 TRINO_IDENTITY_MODE 指定的模式，并拒绝未知模式：不会回退到 service）。向 Trino 发送调用者令牌的模式必须先获取适用于 Trino 的令牌，绝不能转发调用者的令牌：JwtIdentity.trino_token(bearer) 是对应的钩子。若插件要以调用者身份访问自己的后端，则应通过 akko_mcp_trino.identity.current_bearer() 读取已验证的 bearer 令牌，并采用相同方式进行令牌交换，而不是直接发送该令牌。调用者的身份能读取什么，由 Trino 决定。

请求处理流程

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 = this server, exp) → Principal(subject)
    alt no valid token (none, forged, expired, issued for Trino)
        G-->>H: 401  X-Reason: unauthenticated, what to send
    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
    opt jwt mode, no valid exchanged token in memory
        T->>T: exchange the caller's token at the provider (RFC 8693)
    end
    T->>Tr: SQL as the caller (jwt: exchanged token; impersonation: server account), 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

配置

所有内容都从环境变量读取，不会硬编码。扩展也从相同环境中读取自己的变量。

Trino

变量	含义	默认值
TRINO_HOST, TRINO_PORT	协调器	localhost, 8080
TRINO_USER	服务端自身的 Trino 账户：在 service 模式下，每个查询都以该账户运行；在 impersonation 模式下，该账户用于认证并模拟调用者	trino
TRINO_PASSWORD	该账户的密码，通过 Basic 认证发送；在 http 上会被拒绝；--check 会说明是否已设置，但绝不会显示密码	无
TRINO_CLIENT_CERT, TRINO_CLIENT_KEY	该账户用于认证的客户端证书及其密钥（PEM 路径），用于替代密码（mTLS）；仅限 https	无
TRINO_CATALOG	会话目录；读取未指定目录的名称时使用；将其设为数据目录（若为 system，此类名称会被拒绝）	system
TRINO_READ_ONLY	在 execute_query 中拒绝写入	true
TRINO_MAX_ROWS	结果上限，最少为 1；会额外读取一行，以判断答案是否被 truncated	100
TRINO_QUERY_TIMEOUT_SECONDS	单个查询的时间预算：超时后，服务端会在 Trino 中取消查询；0 表示由 Trino 限制	60
MCP_CALL_TIMEOUT_SECONDS	单次工具调用的时间预算，包括该调用中的所有查询：超时后，不会启动新查询，正在运行的查询会被取消，Agent 会收到错误；0 表示不设限制	120
TRINO_SOURCE	Trino 为每个查询记录的来源（X-Trino-Source）	服务端名称
TRINO_HTTP_SCHEME	http 或 https；绝不会通过 http 发送密码	http
TRINO_VERIFY	TLS 验证：true、false，或用于验证 Trino TLS 证书链的 CA bundle 路径	true
TRINO_REQUEST_TIMEOUT_SECONDS	Trino 客户端的单次请求超时时间	30
TRINO_IDENTITY_MODE	服务端在 Trino 中所代表的身份：jwt（每个查询都以调用者身份运行，并通过令牌交换获取 Trino 令牌）、impersonation（每个查询都以调用者身份运行，由 TRINO_USER 进行模拟）或 service（每个查询都以 TRINO_USER 身份运行，并会发出警告）；前两种模式需要 https 和要求有效令牌的入口；扩展可添加其他模式，未知模式会导致启动被拒绝	jwt
TRINO_TOKEN_EXCHANGE_URL	jwt：提供方的令牌端点（仅限 https）	无（jwt 必需）
TRINO_TOKEN_EXCHANGE_CLIENT_ID, TRINO_TOKEN_EXCHANGE_CLIENT_SECRET	jwt：此服务端在提供方的机密客户端凭据，通过 Basic 认证发送；--check 绝不会打印密钥	无（jwt 必需）
TRINO_TOKEN_EXCHANGE_AUDIENCE	jwt：为 Trino 请求的令牌受众，即 Trino 的 required-audience 所指定的受众；绝不能是此服务端接受的受众	无（jwt 必需）
TRINO_TOKEN_EXCHANGE_SCOPE	jwt：要随令牌请求的作用域，以空格分隔	无
TRINO_TOKEN_EXCHANGE_VERIFY	令牌端点的 TLS 验证：true、false，或 CA bundle 路径	true

身份验证

变量	含义	默认值
MCP_AUTH_ENABLED	验证调用方的 bearer 令牌（门禁）	true
MCP_AUTH_REQUIRED	没有有效令牌的请求将被拒绝，并返回 401；jwt 和 impersonation 模式在没有令牌时会拒绝启动	true
MCP_JWKS_URL	提供方的 JWKS 端点	无（启用身份验证时必填）
MCP_OIDC_ISSUER	令牌必须携带的颁发者	无
MCP_OIDC_AUDIENCE	令牌必须携带的受众：提供方为此服务颁发令牌时指定的受众（RFC 8707）；绝不能是 Trino 的受众	无（启用身份验证时必须设置此项或 MCP_RESOURCE_URL）
MCP_RESOURCE_URL	此服务的规范 URL，例如 https://mcp.example.com/mcp：可作为受众，会列入服务随后发布的受保护资源元数据（RFC 9728）以及 401 响应的质询中	无
MCP_JWT_LEEWAY_SECONDS	对 exp/nbf/iat 的容差，用于应对颁发者与此服务之间的时钟偏差	30
MCP_ALLOWED_HOSTS	Host 标头过滤器（SDK 的 DNS 重绑定防护）：以逗号分隔的 host 或 host:port 条目；host:* 表示允许任意端口（唯一支持的通配符）；启用过滤器后，携带 Origin 标头的请求将被拒绝（403）；留空则接受任何 Host	无（关闭）

启用身份验证却未设置 JWKS URL，或未设置此服务自身的受众时，服务会拒绝启动：无法验证任何内容的身份验证层不得假装能够验证。同样，如果门禁要求提供无法验证的令牌，也会拒绝启动（MCP_AUTH_REQUIRED=true 且 MCP_AUTH_ENABLED=false）。

插件和扩展

变量	含义	默认值
MCP_TOOL_PLUGINS	以逗号分隔的注册器名称，这些注册器安装在入口点组 akko_mcp_trino.tools 下	无
MCP_EXTENSIONS	以逗号分隔的扩展名称，这些扩展安装在入口点组 akko_mcp_trino.extensions 下	无

服务

变量	含义	默认值
MCP_TRANSPORT	streamable-http、sse 或 stdio	streamable-http
MCP_PORT、MCP_HEALTH_PORT	监听端口	3000、3001
MCP_SERVER_NAME	向主机公布的名称	trino-mcp
MCP_USER_TOKEN	仅限 stdio：用户的访问令牌，按 bearer 标头的方式验证	无（严格模式下必填）

运维

拒绝请求

每次拒绝都会附带 X-Reason 和请求 ID，绝不会包含令牌。对于没有有效令牌的请求（令牌缺失、伪造、过期或签发给其他受众，包括 Trino 的受众），核心会以 401 unauthenticated 拒绝请求，并在响应正文中说明应发送什么：

{"error": "unauthenticated", "message": "this server requires a bearer token issued for it by the identity provider: send Authorization: Bearer <access token>"}

设置 MCP_RESOURCE_URL 后，服务会在 /.well-known/oauth-protected-resource（以及 URL 路径下，例如 /.well-known/oauth-protected-resource/mcp）发布其受保护资源元数据，注明资源以及作为授权服务器的 MCP_OIDC_ISSUER；按照 MCP 授权规范的要求，401 响应会携带 WWW-Authenticate: Bearer resource_metadata="…"。扩展的防护会添加各自的拒绝原因；其文档和质询优先于核心的文档和质询。

警告

如果配置导致某些内容处于开放状态，系统会在三个地方提示：启动时的一行 WARNING、--check 命令，以及初始化时发送给每个 MCP 主机的 instructions。核心会针对 service 模式以及不要求有效令牌的门禁发出警告。

审计

MCP 端点的每个响应都带有 X-Request-Id。如果网关发送的请求 ID 是纯令牌（仅含字母、数字、.、_、:、-，且最多 128 个字符），就使用该 ID；否则由服务端生成。HTTP 请求之外（stdio）的工具调用会获得自己的 ID。每次工具调用（包括插件的调用）都会向 mcp.audit 日志记录器写入一行以该 ID 为键的 JSON：

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

使用默认日志配置时，该行会带有 INFO:mcp.audit: 前缀；需要纯 JSON 的采集器会去掉此前缀。

字段	含义
request_id、tool、subject	哪次调用、哪个工具、为谁调用；该行没有可用于保存 bearer 或 SQL 文本的字段
ok、error	调用结果；error 是代码而非消息，因为 Trino 的消息可能会引用从某行读取的值（Cannot cast '<value>' to INT）：Trino 错误会记录其错误类型和名称（USER_ERROR:INVALID_CAST_ARGUMENT）；其他异常会记录其类名（TimeoutError、调用超出其预算时的 CallTimeout）；工具为自身拒绝操作指定的代码（read_only、invalid_argument），或插件通过 note(error=...) 指定的代码；否则为 tool_error。Agent 会读取完整消息
duration_ms	调用耗时

调用执行的操作（SQL 的 SHA-256、表、行数以及是否被截断）会在调用运行期间保存在其记录中（akko_mcp_trino.audit.note）；核心不会将其写入日志行，但扩展可以（wrap_audit）。

健康状态与指标

健康检查端口上的 /health 不会访问 Trino，因此 Trino 响应缓慢不会导致存活探针终止服务。/ready 会根据身份模式执行检查：在 jwt 模式下访问 Trino 的公共 /v1/info；在 impersonation 和 service 模式下，以 TRINO_USER 执行有时限的 SELECT 1，不模拟任何人。/metrics 会暴露 mcp_trino_queries_total、mcp_trino_query_errors_total 和 mcp_trino_query_duration_seconds。

开发

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                        # 377 tests; line or branch coverage below 100 % fails the run
integration/run.sh            # live: Keycloak, Trino and three servers in Docker, then removed
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 列出了核心包含的内容（工具、模块、设置、审计字段），并会在新增内容未列入清单时失败。

akko_mcp_trino/ 中的每个模块各司其职，并在 tests/ 中有对应测试：

模块	职责
config.py	读取环境变量，提供中性默认值
auth.py	根据 JWKS 验证 JWT，提供 Principal
identity.py	在 ContextVar 中保存已验证的调用方
middleware.py	防护层：令牌、请求 ID、扩展防护和文档
extensions.py	扩展点及其加载
transport_security.py	Host 标头过滤器（MCP_ALLOWED_HOSTS）
audit.py	审计日志行、请求 ID、记录和调用的时间预算
sql_guard.py	标识符验证，以及基于 AST 的只读判定
tools.py、plugins.py	六个工具和 ToolContext，以及工具插件的加载
trino_client.py	Trino 连接、身份模式（核心中的 jwt、impersonation、service）和时间预算
token_exchange.py	OAuth 2.0 令牌交换（RFC 8693）：用调用方的令牌换取用于 Trino 的令牌
health.py、metrics.py	/health、/ready 和 /metrics；Prometheus 计数器
server.py、app.py、__main__.py	组装、传输、入口点和 --check

README 也会经过测试：此处列出的每个变量都会由 config.py 读取，此处注明的默认值也与代码中的默认值一致。

设计说明

守卫是纯 ASGI 中间件，而不是 BaseHTTPMiddleware。 后者会在单独的任务中运行下游，从而破坏 ContextVar 传播：已验证的调用方身份永远无法传递到工具。

挂载与传输方式无关。 无论使用哪种传输方式，都通过同一代码路径构建传输应用并挂载守卫。守卫曾只存在于默认传输永远不会走到的分支中；现在有测试会断言两种传输方式都包含守卫。

每个工具都会经过身份模式，而不只是 execute_query。 列出目录、架构和表属于读取元数据；有一项测试会遍历每个工具，检查它们是否以同样的方式访问 Trino。

只读策略通过检查 AST 中的所有节点来判定。 只检查开头的关键字会放过 WITH w AS (DELETE FROM t) SELECT 1。只检查根节点也曾如此，直到一次实时对抗性验证发现了问题。现在，守卫会遍历整个语法树。

拒绝表函数。 TABLE(pg.system.query(query => '...')) 会被解析为普通的 SELECT，但它会以连接器自己的账户向数据源发送原始 SQL，因此引擎不会应用行过滤器或列掩码。守卫会拒绝所有表函数以及 system 目录（其中的 system.runtime.queries 会显示其他用户的 SQL）。引擎策略也必须拒绝它们；守卫是第二道防线，而不是唯一防线。

不透传令牌。 在 0.5 及之前，jwt 模式会在验证调用方令牌后，将其发送给 Trino。MCP 授权规范禁止这样做：为一个服务签发的令牌不得继续发送给另一个服务，否则该服务可能会信任一个原本并非发给它的令牌（即“混淆代理”问题）。从 0.6 起，服务端只接受受众为自身的令牌，拒绝发给 Trino 的令牌，并通过 OAuth 2.0 Token Exchange（RFC 8693）为 Trino 获取专用令牌。如果提供方无法交换令牌，则显式使用 impersonation 作为回退方式；绝不会默默选择。

始终使用经过验证的身份，而非调用方指定的身份。 调用方的名称始终来自其经过验证的令牌。调用方发送一个指定 Trino 用户的标头不会产生任何影响：服务端会自行构建发往 Trino 的请求。密码、令牌或客户端证书绝不会通过普通 http 发送；如果核心不认识某种身份模式，服务会拒绝启动，而不会回退到其他模式。

默认安全。 在 0.5 及之前，默认情况下所有查询都以 TRINO_USER 身份运行，且没有令牌的请求也会得到处理。从 0.6 起，默认情况下所有查询都以调用方身份运行，并拒绝没有有效令牌的请求；仍可按名称选择 service 模式和开放访问，但会发出警告。行数和时间预算也都有默认值：无限制曾是一种无人选择的设置。

扩展只能拒绝请求。 扩展守卫会在令牌验证期间运行，而不是取代令牌验证；无论它们给出什么结果，只读守卫都会照常应用。

服务端不负责决定访问权限。 这由 Trino 针对每个调用方或配置的身份来决定。在服务端设置策略会造成重复，并最终与引擎对其他所有客户端实施的策略相矛盾。

贡献

欢迎贡献：请参阅 CONTRIBUTING.md 了解基本规则（先写测试、100 % 覆盖率、产品中立、默认拒绝），并参阅 SECURITY.md 了解如何私下报告漏洞。变更记录见 CHANGELOG.md。

关于 AKKO

akko-mcp-trino 由 AKKO 构建并维护。AKKO 是一个法国项目，致力于为 AI Agent 提供受治理的企业数据访问能力；数据保留在原处，并使用组织已经运行的引擎和身份提供方。

维护者： AKKO（[email protected]） · 问题反馈： github.com/AKKO-p/akko-mcp-trino/issues

许可证

Apache 2.0。请参阅 LICENSE。