trino_mcp
项目介绍
Trino MCP 服务器
Trino 的 Model Context Protocol 服务器，为 AI 模型提供对 Trino 分布式 SQL 查询引擎的结构化访问。
⚠️ 测试版 (v0.1.2) ⚠️
该项目的核心功能已经稳定并经过测试。欢迎 fork 并贡献！
功能
- ✅ 解决了 Docker 容器 API 初始化问题！（可靠的服务器初始化）
- ✅ 通过 MCP 协议暴露 Trino 资源
- ✅ 允许 AI 工具查询和分析 Trino 中的数据
- ✅ 提供传输选项（STDIO 传输可靠；SSE 传输存在问题）
- ✅ 解决了目录处理问题，以确保正确的 Trino 查询执行
- ✅ 提供 Docker 容器 API 和独立的 Python API 服务器选项
快速开始
# 使用 docker-compose 启动服务器
docker-compose up -d
# 验证 API 是否工作
curl -X POST "http://localhost:9097/api/query" \
-H "Content-Type: application/json" \
-d '{"query": "SELECT 1 AS test"}'
需要非容器化的版本？运行独立的 API：
# 在端口 8008 上运行独立的 API 服务器
python llm_trino_api.py
LLM 集成
希望给 LLM 直接访问查询你的 Trino 实例的能力吗？我们创建了一些简单的工具来实现这一点！
命令行 LLM 接口
让 LLM 查询 Trino 的最简单方法是通过我们的命令行工具：
# 简单直接查询（非常适合 LLM）
python llm_query_trino.py "SELECT * FROM memory.bullshit.real_bullshit_data LIMIT 5"
# 指定不同的目录或模式
python llm_query_trino.py "SELECT * FROM information_schema.tables" memory information_schema
LLM 的 REST API
我们提供了两种 API 选项用于与 LLM 应用程序集成：
1. Docker 容器 API（端口 9097）
Docker 容器在端口 9097 上暴露一个 REST API：
# 对 Docker 容器 API 执行查询
curl -X POST "http://localhost:9097/api/query" \
-H "Content-Type: application/json" \
-d '{"query": "SELECT 1 AS test"}'
2. 独立的 Python API（端口 8008）
为了更灵活的部署，运行独立的 API 服务器：
# 在端口 8008 上启动 API 服务器
python llm_trino_api.py
这会在以下位置创建端点：
GET http://localhost:8008/- API 使用信息POST http://localhost:8008/query- 执行 SQL 查询
然后可以让 LLM 向此端点发出 HTTP 请求：
# 示例代码，LLM 可能生成
import requests
def query_trino(sql_query):
response = requests.post(
"http://localhost:8008/query",
json={"query": sql_query}
)
return response.json()
# LLM 生成的查询
results = query_trino("SELECT job_title, AVG(salary) FROM memory.bullshit.real_bullshit_data GROUP BY job_title ORDER BY AVG(salary) DESC LIMIT 5")
print(results["formatted_results"])
这种方法允许 LLM 专注于生成 SQL，而我们的工具处理所有 MCP 协议复杂性！
演示和验证脚本 🚀
我们创建了一些强大的演示脚本，展示了 AI 模型如何使用 MCP 协议对 Trino 运行复杂的查询：
1. 虚假数据生成和加载
tools/create_bullshit_data.py 脚本生成一个包含 10,000 名员工的数据集，这些员工具有荒谬的工作头衔、虚高的薪水以及“虚假因素”评分（1-10）：
# 生成虚假数据
python tools/create_bullshit_data.py
# 将虚假数据加载到 Trino 的内存目录中
python load_bullshit_data.py
2. 通过 MCP 运行复杂查询
test_bullshit_query.py 脚本演示了端到端的 MCP 交互：
- 使用 STDIO 传输连接到 MCP 服务器
- 根据 MCP 规范初始化协议
- 运行带有 WHERE、GROUP BY、HAVING、ORDER BY 的复杂 SQL 查询
- 处理并格式化结果
# 通过 MCP 对虚假数据运行复杂查询
python test_bullshit_query.py
示例输出显示薪水高且虚假因素高的顶级职位：
🏆 TOP 10 BULLSHIT JOBS (high salary, high BS factor):
----------------------------------------------------------------------------------------------------
JOB_TITLE | COUNT | AVG_SALARY | MAX_SALARY | AVG_BS_FACTOR
----------------------------------------------------------------------------------------------------
Advanced Innovation Jedi | 2 | 241178.50 | 243458.00 | 7.50
VP of Digital Officer | 1 | 235384.00 | 235384.00 | 7.00
Innovation Technical Architect | 1 | 235210.00 | 235210.00 | 9.00
...and more!
3. API 测试
test_llm_api.py 脚本验证 API 功能：
# 测试 Docker 容器 API
python test_llm_api.py
这会进行全面检查：
- API 端点发现
- 文档可用性
- 有效查询执行
- 错误处理无效查询
使用
# 使用 docker-compose 启动服务器
docker-compose up -d
服务器将在以下位置可用：
- Trino: http://localhost:9095
- MCP 服务器: http://localhost:9096
- API 服务器: http://localhost:9097
客户端连接
✅ 重要：客户端脚本在您的本地机器上运行（位于 Docker 外部），并连接到 Docker 容器。脚本自动通过使用 docker exec 命令处理这一点。您不需要进入容器即可使用 MCP！
从本地机器运行测试：
# 生成并将数据加载到 Trino
python tools/create_bullshit_data.py # 本地生成数据
python load_bullshit_data.py # 将数据加载到 Docker 中的 Trino
# 通过 Docker 运行 MCP 查询
python test_bullshit_query.py # 使用 Docker 中的 MCP 进行查询
传输选项
此服务器支持两种传输方法，但目前只有 STDIO 是可靠的：
STDIO 传输（推荐且工作正常）
STDIO 传输工作可靠，并且目前是唯一推荐的方法进行测试和开发：
# 在容器内使用 STDIO 传输运行
docker exec -i trino_mcp_trino-mcp_1 python -m trino_mcp.server --transport stdio --debug --trino-host trino --trino-port 8080 --trino-user trino --trino-catalog memory
SSE 传输（不推荐 - 存在关键问题）
SSE 是 MCP 的默认传输方式，但在当前 MCP 1.3.0 版本中存在严重问题，导致客户端断开连接时服务器崩溃。直到这些问题解决之前不推荐使用：
# 不推荐：使用 SSE 传输运行（断开连接时崩溃）
docker exec trino_mcp_trino-mcp_1 python -m trino_mcp.server --transport sse --host 0.0.0.0 --port 8000 --debug
已知问题及修复
固定：Docker 容器 API 初始化
✅ 已修复：我们解决了 Docker 容器中的 API 返回 503 服务不可用响应的问题。问题在于 app_lifespan 函数没有正确初始化 app_context_global 和 Trino 客户端连接。修复确保：
- Trino 客户端在启动时显式连接
- AppContext 全局变量正确初始化
- 健康检查现在正常工作
如果遇到 503 错误，请检查容器是否使用最新代码重建：
# 使用修复重新构建并重启容器
docker-compose stop trino-mcp
docker-compose rm -f trino-mcp
docker-compose up -d trino-mcp
MCP 1.3.0 SSE 传输崩溃
MCP 1.3.0 的 SSE 传输存在关键问题，导致客户端断开连接时服务器崩溃。直到集成更新的 MCP 版本之前，仅使用 STDIO 传输。错误表现为：
RuntimeError: generator didn't stop after athrow()
anyio.BrokenResourceError
Trino 目录处理
我们修复了 Trino 客户端的目录处理问题。原始实现尝试使用 USE catalog 语句，但这不能可靠地工作。修复直接在连接参数中设置目录。
项目结构
此项目组织如下：
src/- Trino MCP 服务器的主要源代码examples/- 显示如何使用服务器的简单示例scripts/- 有用的诊断和测试脚本tools/- 数据创建和设置的实用脚本tests/- 自动化测试
关键文件：
llm_trino_api.py- 用于 LLM 集成的独立 API 服务器test_llm_api.py- API 服务器的测试脚本test_mcp_stdio.py- 主要测试脚本，使用 STDIO 传输（推荐）test_bullshit_query.py- 使用虚假数据的复杂查询示例load_bullshit_data.py- 将生成的数据加载到 Trino 的脚本tools/create_bullshit_data.py- 生成搞笑测试数据的脚本run_tests.sh- 运行自动化测试的脚本examples/simple_mcp_query.py- 使用 MCP 查询数据的简单示例
开发
重要：所有脚本都可以从您的本地机器运行 - 它们会自动通过 docker exec 命令与 Docker 容器通信！
# 安装开发依赖
pip install -e ".[dev]"
# 运行自动化测试
./run_tests.sh
# 使用 STDIO 传输测试 MCP（推荐）
python test_mcp_stdio.py
# 简单示例查询
python examples/simple_mcp_query.py "SELECT 'Hello World' AS message"
测试
要测试 Trino 查询是否正确工作，请使用 STDIO 传输测试脚本：
# 推荐的测试方法（STDIO 传输）
python test_mcp_stdio.py
对于使用虚假数据的更复杂测试：
# 加载并查询虚假数据（展示 Trino MCP 的全部威力！）
python load_bullshit_data.py
python test_bullshit_query.py
对于测试 LLM API 端点：
# 测试 Docker 容器 API
python test_llm_api.py
# 测试独立 API（确保它先运行）
python llm_trino_api.py
curl -X POST "http://localhost:8008/query" \
-H "Content-Type: application/json" \
-d '{"query": "SELECT 1 AS test"}'
LLM 如何使用这个
LLM 可以使用 Trino MCP 服务器来：
-
获取数据库模式信息：
# 示例提示给 LLM：“memory 目录中有哪些模式？” # LLM 可以生成查询代码： query = "SHOW SCHEMAS FROM memory" -
运行复杂的分析查询：
# 示例提示：“找出平均薪资最高的前五个职位” # LLM 可以生成复杂的 SQL： query = """ SELECT job_title, AVG(salary) as avg_salary FROM memory.bullshit.real_bullshit_data GROUP BY job_title ORDER BY avg_salary DESC LIMIT 5 """ -
执行数据分析并呈现结果：
# LLM 可以解析响应，提取见解并向用户呈现： "最高薪酬的职位是 'Advanced Innovation Jedi'，平均薪酬为 $241,178.50"
真实 LLM 分析示例：公司中的虚假职位
这里是一个真实的示例，当被问及“识别拥有最多虚假职位员工的公司并创建 Mermaid 图表”时，LLM 可能会产生什么：
步骤 1：LLM 生成并运行查询
SELECT
company,
COUNT(*) as employee_count,
AVG(bullshit_factor) as avg_bs_factor
FROM
memory.bullshit.real_bullshit_data
WHERE
bullshit_factor > 7
GROUP BY
company
ORDER BY
employee_count DESC,
avg_bs_factor DESC
LIMIT 10
步骤 2：LLM 获取并分析结果
COMPANY | EMPLOYEE_COUNT | AVG_BS_FACTOR
----------------------------------------
Unknown Co | 2 | 9.0
BitEdge | 1 | 10.0
CyberWare | 1 | 10.0
BitLink | 1 | 10.0
AlgoMatrix | 1 | 10.0
CryptoHub | 1 | 10.0
BitGrid | 1 | 10.0
MLStream | 1 | 10.0
CloudCube | 1 | 10.0
UltraEdge | 1 | 10.0
步骤 3：LLM 生成 Mermaid 图表可视化
%%{init: {'theme': 'forest'}}%%
graph LR
title[Companies with Most Bullshit Jobs]
style title fill:#333,stroke:#333,stroke-width:1px,color:white,font-weight:bold,font-size:18px
Companies --> UnknownCo[Unknown Co]
Companies --> BitEdge[BitEdge]
Companies --> CyberWare[CyberWare]
Companies --> BitLink[BitLink]
Companies --> AlgoMatrix[AlgoMatrix]
Companies --> CryptoHub[CryptoHub]
Companies --> BitGrid[BitGrid]
Companies --> MLStream[MLStream]
Companies --> CloudCube[CloudCube]
Companies --> UltraEdge[UltraEdge]
UnknownCo --- Count2[2 employees]
BitEdge --- Count1a[1 employee]
CyberWare --- Count1b[1 employee]
BitLink --- Count1c[1 employee]
AlgoMatrix --- Count1d[1 employee]
CryptoHub --- Count1e[1 employee]
BitGrid --- Count1f[1 employee]
MLStream --- Count1g[1 employee]
CloudCube --- Count1h[1 employee]
UltraEdge --- Count1i[1 employee]
classDef company fill:#ff5733,stroke:#333,stroke-width:1px,color:white,font-weight:bold;
classDef count fill:#006100,stroke:#333,stroke-width:1px,color:white,font-weight:bold;
class UnknownCo,BitEdge,CyberWare,BitLink,AlgoMatrix,CryptoHub,BitGrid,MLStream,CloudCube,UltraEdge company;
class Count2,Count1a,Count1b,Count1c,Count1d,Count1e,Count1f,Count1g,Count1h,Count1i count;
替代条形图：
%%{init: {'theme': 'default'}}%%
pie showData
title Companies with Bullshit Jobs
"Unknown Co (BS: 9.0)" : 2
"BitEdge (BS: 10.0)" : 1
"CyberWare (BS: 10.0)" : 1
"BitLink (BS: 10.0)" : 1
"AlgoMatrix (BS: 10.0)" : 1
"CryptoHub (BS: 10.0)" : 1
"BitGrid (BS: 10.0)" : 1
"MLStream (BS: 10.0)" : 1
"CloudCube (BS: 10.0)" : 1
"UltraEdge (BS: 10.0)" : 1
步骤 4：LLM 提供关键见解
LLM 可以分析数据并提供见解：
- “Unknown Co”有最多的虚假职位员工（2人），而其他公司只有1人
- 大多数公司达到了完美的10.0虚假因素评分
- 技术导向的公司（如 BitEdge、CyberWare 等）似乎创造了特别无意义的角色
- 虚假角色集中在高管或专业职位级别
这个示例展示了 LLM 如何：
- 根据自然语言问题生成适当的 SQL 查询
- 处理并解释来自 Trino 的结果
- 创建数据的可视化表示
- 提供有意义的见解和分析
访问 API
Trino MCP 服务器现在包括两个 API 选项用于访问数据：
1. Docker 容器 API（端口 9097）
import requests
import json
# API 端点（默认端口 9097 在 Docker 设置中）
api_url = "http://localhost:9097/api/query"
# 定义您的 SQL 查询
query_data = {
"query": "SELECT * FROM memory.bullshit.real_bullshit_data LIMIT 5",
"catalog": "memory",
"schema": "bullshit"
}
# 发送请求
response = requests.post(api_url, json