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 服务器选项

快速开始

# 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

我们提供了两种 API 选项，用于与 LLM 应用程序集成：

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

服务器将在以下地址可用：

• Trino: http://localhost:9095
• MCP 服务器: http://localhost:9096
• API 服务器: http://localhost:9097

客户端连接

✅ 重要提示: 客户端脚本在您的本地机器上运行（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 客户端连接。修复确保了：

1. Trino 客户端在启动时显式连接
2. AppContext 全局变量被正确初始化
3. 健康检查现在可以正常工作

如果您遇到 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

对于更复杂的测试，可以使用 bullshit 数据：

# 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 如何使用这个

LLM 可以使用 Trino MCP 服务器来：

1. 获取数据库模式信息：

   # 示例提示给 LLM: "memory 目录中有哪些模式可用？"
   # LLM 可以生成查询代码：
   query = "SHOW SCHEMAS FROM memory"
   

2. 运行复杂的分析查询：

   # 示例提示: "找出平均薪资最高的前5个职位"
   # 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
   """
   

3. 执行数据分析并呈现结果：

   # LLM 可以解析响应，提取见解并向用户展示：
   "薪资最高的职位是 'Advanced Innovation Jedi'，平均薪资为 $241,178.50"
   

实际 LLM 分析示例：按公司划分的无意义工作

这是一个实际的例子，当要求 LLM “识别拥有最多无意义工作岗位员工的公司，并创建一个 Mermaid 图表”时，LLM 可以生成的内容：

第一步：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

第三步：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

第四步：LLM 提供关键见解

LLM 可以分析数据并提供见解：

• "Unknown Co" 在无意义角色中的员工最多（2名），而其他公司都只有1名
• 大多数公司达到了完美的 10.0 无意义因素评分
• 专注于技术的公司（如 BitEdge、CyberWare 等）似乎创造了特别没有意义的角色
• 无意义的角色似乎集中在高管或专业职位级别

这个例子展示了 LLM 如何：

1. 根据自然语言问题生成适当的 SQL 查询
2. 处理和解释来自 Trino 的结果
3. 创建数据的可视化表示
4. 提供有意义的见解和分析

访问 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 错误：

1. 确保你已经用最新代码重建了容器：

   docker-compose stop trino-mcp
   docker-compose rm -f trino-mcp
   docker-compose up -d trino-mcp
   

2. 检查容器日志中的错误：

   docker logs trino_mcp_trino-mcp_1
   

3. 验证 Trino 是否正常运行：

   curl -s http://localhost:9095/v1/info | jq
   

与独立 API 的端口冲突

独立 API 默认使用 8008 端口以避免冲突。如果你看到“地址已被使用”的错误：

1. 编辑 llm_trino_api.py 并在最后一行更改端口号：

   uvicorn.run(app, host="127.0.0.1", port=8008) 
   

2. 通过命令行使用自定义端口运行：

   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年