Trino MCP 服务器

Trino 的模型上下文协议服务器，为 AI 模型提供对 Trino 分布式 SQL 查询引擎的结构化访问。

⚠️ BETA 发布 (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

显示高薪的顶级垃圾工作的示例输出：

🏆 顶级 10 大垃圾工作（高薪，高垃圾因子）：
----------------------------------------------------------------------------------------------------
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
...还有更多！

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 Service Unavailable 响应的问题。问题在于 app_lifespan 函数没有正确初始化 app_context_global 和 Trino 客户端连接。此修复确保：

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

如果您遇到 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 服务器来：

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 可以生成什么：

步骤 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 生成美人鱼图表可视化

%%{init: {'theme': 'forest'}}%%
graph LR
    title[拥有最多垃圾工作的公司]
    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)，而所有其他公司只有一名
• 大多数公司都获得了完美的 10.0 垃圾因子评分
• 以技术为中心的公司（BitEdge、CyberWare 等）似乎创造了特别没有意义的角色
• 垃圾角色似乎集中在执行或专业职位级别

此示例演示了 LLM 如何：

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

访问 API

Trino MCP 服务器现在包含两个用于访问数据的 API 选项：

1. Docker 容器 API (端口 9097)

import requests
import json

# API 端点（Docker 设置中的默认端口 9097）
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=query_data)
results = response.json()

# 处理结果
if results["success"]:
    print(f"查询返回 {results['results']['row_count']} 行")
    for row in results['results']['rows']:
        print(row)
else:
    print(f"查询失败：{results['message']}")

2. 独立的 Python API (端口 8008)

# 与上述代码相同，但端口不同
api_url = "http://localhost:8008/query"

两个 API 都提供以下端点：

• GET /api - API 文档和使用示例
• POST /api/query - 针对 Trino 执行 SQL 查询

这些 API 消除了对包装器脚本的需求，并允许 LLM 使用 REST 调用直接查询 Trino，从而使其更容易与 Claude、GPT 和其他 AI 系统集成。

故障排除

API 返回 503 Service Unavailable

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

未来工作

现在处于 beta 阶段，计划进行以下改进：

• [ ] 在可用的情况下与较新的 MCP 版本集成，以修复 SSE 传输问题
• [ ] 添加/验证对 Hive、JDBC 和其他连接器的支持
• [ ] 跨不同类型和复杂性添加更全面的查询验证
• [ ] 实现对更多数据类型和高级 Trino 功能的支持
• [ ] 改进错误处理和恢复机制
• [ ] 添加用户身份验证和权限控制
• [ ] 创建更全面的示例和文档
• [ ] 开发管理监控和管理界面
• [ ] 添加性能指标和查询优化提示
• [ ] 实现对长时间运行的查询和结果流的支持

---

由 Stink Labs 开发，2025 年