Trino MCP Server

Trino MCP 服务

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

⚠️ BETA RELEASE (v0.1.2) ⚠️
此项目正在趋于稳定，核心功能已正常运行并通过测试。欢迎派生项目并贡献代码！

功能

• ✅ 已修复 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

大模型集成

想让大模型直接访问并查询你的 Trino 实例？我们为此创建了简单的工具！

命令行大模型接口

让大模型查询 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

面向大模型的 REST API

我们提供两种 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 查询

然后，你可以让大模型向此端点发送 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"])

这种方式让大模型专注于生成 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

展示高薪荒唐职位排行的示例输出：

🏆 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

客户端连接

✅ IMPORTANT：客户端脚本在你的本地计算机上运行（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 Service Unavailable 响应的问题。问题在于 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 Catalog 处理

我们修复了 Trino 客户端中的 Catalog 处理问题。原始实现尝试使用 USE catalog 语句，但这种方式不可靠。修复后，连接参数会直接设置 Catalog。

项目结构

本项目的组织结构如下：

• src/ - Trino MCP 服务的主要源代码
• examples/ - 展示如何使用服务的简单示例
• scripts/ - 实用的诊断和测试脚本
• tools/ - 用于创建数据和设置环境的工具脚本
• tests/ - 自动化测试

关键文件：

• llm_trino_api.py - 用于集成大模型的独立 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

测试大模型 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"}'

大模型如何使用此服务

大模型可以使用 Trino MCP 服务来：

1. 获取数据库架构信息：

   # Example prompt to LLM: "What schemas are available in the memory catalog?"
   # LLM can generate code to query:
   query = "SHOW SCHEMAS FROM memory"
   

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

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

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

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

大模型分析实例：按公司统计 Bullshit Jobs 数量

以下是一个实际示例，展示了大模型在收到“找出从事 Bullshit Jobs 的员工最多的公司，并创建一个 Mermaid 图表”这一请求时可能生成的内容：

第 1 步：大模型生成并运行查询

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 步：大模型获取并分析结果

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 步：大模型生成 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 步：大模型提供关键见解

大模型可以分析数据并提供见解：

• “Unknown Co”的 Bullshit Jobs 员工人数最多（2 人），其他公司都只有 1 人
• 大多数公司的 Bullshit Factor 得分都达到了满分 10.0
• 专注于科技的公司（BitEdge、CyberWare 等）似乎创造了尤其毫无意义的职位
• Bullshit 职位似乎集中在高管或专业岗位层级

此示例展示了大模型如何：

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 无需编写包装脚本，让大模型能够直接通过 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，以避免冲突。如果看到“address already in use”错误：

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