Trino MCP 服务器
Trino 的模型上下文协议服务器，为 AI 模型提供对 Trino 分布式 SQL 查询引擎的结构化访问。
⚠️测试版 (v0.1.2) ⚠️
该项目正在稳定运行，核心功能已测试完毕。欢迎随时 fork 并贡献代码！
特征
- ✅ 修复 Docker 容器 API 初始化问题！（可靠服务器初始化）
- ✅ 通过 MCP 协议公开 Trino 资源
- ✅ 使 AI 工具能够查询和分析 Trino 中的数据
- ✅ 提供传输选项（STDIO 传输可靠；SSE 传输存在问题）
- ✅ 修复了目录处理，以便正确执行 Trino 查询
- ✅ Docker 容器 API 和独立 Python API 服务器选项
快速入门
# Start the server with docker-compose
docker-compose up -d
# Verify the API is working
curl -X POST "http://localhost:9097/api/query" \
-H "Content-Type: application/json" \
-d '{"query": "SELECT 1 AS test"}'
需要非容器化版本？运行独立 API：
# Run the standalone API server on port 8008
python llm_trino_api.py
LLM 整合
想要让法学硕士 (LLM) 直接访问您的 Trino 实例吗？我们为此创建了简单的工具！
命令行 LLM 界面
让 LLM 查询 Trino 的最简单方法是通过我们的命令行工具：
# Simple direct query (perfect for LLMs)
python llm_query_trino.py "SELECT * FROM memory.bullshit.real_bullshit_data LIMIT 5"
# Specify a different catalog or schema
python llm_query_trino.py "SELECT * FROM information_schema.tables" memory information_schema
法学硕士 (LLM) 的 REST API
我们提供两种与 LLM 应用程序集成的 API 选项：
1. Docker 容器 API（端口 9097）
Docker 容器在端口 9097 上公开 REST API：
# Execute a query against the Docker container 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 服务器：
# Start the API server on port 8008
python llm_trino_api.py
这将在以下位置创建端点：
GET http://localhost:8008/ - API 使用信息POST http://localhost:8008/query - 执行 SQL 查询
然后，您可以让您的 LLM 向此端点发出 HTTP 请求：
# Example code an LLM might generate
import requests
def query_trino(sql_query):
response = requests.post(
"http://localhost:8008/query",
json={"query": sql_query}
)
return response.json()
# LLM-generated query
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）：
# Generate the bullshit data
python tools/create_bullshit_data.py
# Load the bullshit data into Trino's memory catalog
python load_bullshit_data.py
2. 通过 MCP 运行复杂查询
test_bullshit_query.py脚本演示了端到端 MCP 交互：
- 使用 STDIO 传输连接到 MCP 服务器
- 按照 MCP 规范初始化协议
- 使用 WHERE、GROUP BY、HAVING、ORDER BY 运行复杂的 SQL 查询
- 处理并格式化结果
# Run a complex query against the bullshit data through MCP
python test_bullshit_query.py
示例输出显示了薪水最高的 BS 职位：
🏆 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 功能：
# Test the Docker container API
python test_llm_api.py
这将执行以下全面检查：
- API 端点发现
- 文档可用性
- 有效的查询执行
- 无效查询的错误处理
用法
# Start the server with docker-compose
docker-compose up -d
该服务器将在以下位置可用：
客户端连接
✅重要提示：客户端脚本在您的本地机器（Docker 外部）运行，并连接到 Docker 容器。脚本会使用 docker exec 命令自动处理此操作。您无需进入容器即可使用 MCP！
从本地机器运行测试：
# Generate and load data into Trino
python tools/create_bullshit_data.py # Generates data locally
python load_bullshit_data.py # Loads data to Trino in Docker
# Run MCP query through Docker
python test_bullshit_query.py # Queries using MCP in Docker
交通选择
该服务器支持两种传输方法，但目前只有 STDIO 可靠：
STDIO 传输（推荐且有效）
STDIO 传输工作可靠，是目前唯一推荐的测试和开发方法：
# Run with STDIO transport inside the container
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 版本存在严重问题，导致客户端断开连接时服务器崩溃。在这些问题解决之前，不建议使用：
# NOT RECOMMENDED: Run with SSE transport (crashes on disconnection)
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 错误，请检查您的容器是否已使用最新代码重建：
# Rebuild and restart the container with the fix
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 容器通信！
# Install development dependencies
pip install -e ".[dev]"
# Run automated tests
./run_tests.sh
# Test MCP with STDIO transport (recommended)
python test_mcp_stdio.py
# Simple example query
python examples/simple_mcp_query.py "SELECT 'Hello World' AS message"
测试
要测试 Trino 查询是否正常工作，请使用 STDIO 传输测试脚本：
# Recommended test method (STDIO transport)
python test_mcp_stdio.py
对于使用废话数据进行更复杂的测试：
# Load and query the bullshit data (shows the full power of Trino MCP!)
python load_bullshit_data.py
python test_bullshit_query.py
用于测试 LLM API 端点：
# Test the Docker container API
python test_llm_api.py
# Test the standalone API (make sure it's running first)
python llm_trino_api.py
curl -X POST "http://localhost:8008/query" \
-H "Content-Type: application/json" \
-d '{"query": "SELECT 1 AS test"}'
法学硕士如何利用这一点
LLM 可以使用 Trino MCP 服务器来：
- 获取数据库架构信息：
# Example prompt to LLM: "What schemas are available in the memory catalog?"
# LLM can generate code to query:
query = "SHOW SCHEMAS FROM memory"
- 运行复杂的分析查询：
# Example prompt: "Find the top 5 job titles with highest average salaries"
# LLM can generate complex 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 can parse the response, extract insights and present to user:
"The highest paying job title is 'Advanced Innovation Jedi' with an average salary of $241,178.50"
真实的法学硕士分析示例：公司提供的垃圾工作
以下是一个真实的例子，说明当法学硕士被要求“找出从事垃圾工作最多的员工的公司并创建一个美人鱼图表”时，他们可以得出什么结论：
步骤 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
第二步：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 生成美人鱼图表可视化
替代条形图：
第四步：法学硕士提供关键见解
LLM 可以分析数据并提供见解：
- “未知公司” 的无意义职位员工最多（2 人），而其他所有公司只有 1 人
- 大多数公司的“废话”因素得分都达到了完美的 10.0
- 专注于技术的公司（BitEdge、CyberWare 等）似乎创造了特别无意义的职位
- 垃圾角色似乎集中在高管或专业职位层面
此示例演示了法学硕士 (LLM) 如何能够：
- 根据自然语言问题生成适当的 SQL 查询
- 处理并解释 Trino 的结果
- 创建数据的可视化表示
- 提供有意义的见解和分析
访问 API
Trino MCP 服务器现在包含两个用于访问数据的 API 选项：
1. Docker 容器 API（端口 9097）
import requests
import json
# API endpoint (default port 9097 in Docker setup)
api_url = "http://localhost:9097/api/query"
# Define your SQL query
query_data = {
"query": "SELECT * FROM memory.bullshit.real_bullshit_data LIMIT 5",
"catalog": "memory",
"schema": "bullshit"
}
# Send the request
response = requests.post(api_url, json=query_data)
results = response.json()
# Process the results
if results["success"]:
print(f"Query returned {results['results']['row_count']} rows")
for row in results['results']['rows']:
print(row)
else:
print(f"Query failed: {results['message']}")
2.独立 Python API（端口 8008）
# Same code as above, but with different port
api_url = "http://localhost:8008/query"
这两个 API 都提供以下端点：
GET /api - API 文档和使用示例POST /api/query - 针对 Trino 执行 SQL 查询
这些 API 消除了对包装脚本的需求，并允许 LLM 使用 REST 调用直接查询 Trino，从而使得与 Claude、GPT 和其他 AI 系统等服务的集成变得更加简单。
故障排除
API 返回 503 服务不可用
如果 Docker 容器 API 返回 503 错误：
- 确保您已经使用最新代码重建了容器：
docker-compose stop trino-mcp
docker-compose rm -f trino-mcp
docker-compose up -d trino-mcp
- 检查容器日志中是否存在错误：
docker logs trino_mcp_trino-mcp_1
- 验证 Trino 是否正常运行：
curl -s http://localhost:9095/v1/info | jq
端口与独立 API 冲突
独立 API 默认使用端口 8008，以避免冲突。如果您看到“地址已被使用”错误：
- 编辑
llm_trino_api.py并更改最后一行的端口号：uvicorn.run(app, host="127.0.0.1", port=8008)
- 通过命令行使用自定义端口运行：
python -c "import llm_trino_api; import uvicorn; uvicorn.run(llm_trino_api.app, host='127.0.0.1', port=8009)"
未来工作
目前处于测试阶段，计划进行以下改进：
- [ ] 与较新的 MCP 版本集成（如果可用），以修复 SSE 传输问题
- [ ] 添加/验证对 Hive、JDBC 和其他连接器的支持
- [ ] 添加更全面的跨不同类型和复杂程度的查询验证
- [ ] 实现对更多数据类型和高级 Trino 功能的支持
- [ ] 改进错误处理和恢复机制
- [ ] 添加用户认证和权限控制
- [ ] 创建更全面的示例和文档
- [ ] 开发管理员监控和管理界面
- [ ] 添加性能指标和查询优化提示
- [ ] 实现对长时间运行的查询和结果流的支持
由 Stink Labs 于 2025 年开发