Trino MCP Server

Trino MCP服务器

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

⚠️ 贝塔释放（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

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.码头集装箱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 该脚本生成了一个包含10000名员工的数据集，这些员工拥有荒谬的职位、虚高的工资和“胡说八道因素”评级（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

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

• 三位一体 ：http://localhost:9095
• MCP服务器：http://localhost:9096
• API服务器：http://localhost:9097

客户端连接

✅ 重要客户端脚本在您的本地机器（OUTSIDE 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是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

对于使用废话数据进行更复杂的测试：

# 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. 获取数据库架构信息:

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

1. 运行复杂的分析查询:

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

1. 执行数据分析并呈现结果:

   # 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

步骤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[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

第四步：法学硕士提供关键见解

法学硕士可以分析数据并提供见解：

• “Unknown Co”拥有最多的废话角色员工（2），而其他所有人只有一个
• 大多数公司都取得了完美的10.0废话因子得分
• 以技术为重点的公司（BitEdge、CyberWare等）似乎创造了特别无意义的角色
• 胡说八道的角色似乎集中在高管或专业职位级别

此示例演示了LLM如何：

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

访问API

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

1.码头集装箱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

1. 检查容器日志是否有错误：

   docker logs trino_mcp_trino-mcp_1

1. 验证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) 

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

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