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

该服务器将在以下位置可用：

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 客户端连接造成的。此修复可确保：

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 年开发

You must be authenticated.

security – no known vulnerabilities

license - permissive license

quality - confirmed to work

hybrid server

The server is able to function both locally and remotely, depending on the configuration or use case.

Tools

为 AI 模型提供对 Trino 分布式 SQL 查询引擎的结构化访问，使 LLM 能够直接查询和分析存储在 Trino 数据库中的数据。

Related Resources

Reddit Discussion about this server

Related MCP Servers

MySQL MCP Server
designcomputer
A
security
A
license
A
quality
Allows AI assistants to list tables, read data, and execute SQL queries through a controlled interface, making database exploration and analysis safer and more structured.
Last updated -
1
556
Python
MIT License
Microsoft SQL Server MCP Server (MSSQL)
dperussina
-
security
A
license
-
quality
An easy-to-use bridge that lets AI assistants like Claude and Cursor IDE directly query and explore Microsoft SQL Server databases. No coding experience required!
Last updated -
27
JavaScript
GPL 3.0
Trino MCP Server
Dataring-engineering
-
security
A
license
-
quality
An MCP server that enables listing and querying Trino tables using Python, providing access to data resources through the Model-Control-Protocol.
Last updated -
13
Python
MIT License
Metabase Server MCP
berry-street
-
security
F
license
-
quality
Enables AI assistants to interact with Metabase, providing access to dashboards, questions, databases, and tools for executing queries and viewing data through natural language.
Last updated -
JavaScript

View all related MCP servers

Appeared in Searches

An MCP server that does not require API configuration