Trino MCP Server

Trino MCP サーバー

Trino 向けの Model Context Protocol サーバーです。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

LLM アプリケーションと統合するための API を 2 種類用意しています：

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 スクリプトは、ばかげた役職名、誇張された給与、「でたらめ度」評価（1～10）を持つ従業員 10,000 人分のデータセットを生成します：

# 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

クライアント接続

✅ 重要：クライアントスクリプトはローカルマシン（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

トランスポート方式

このサーバーは 2 種類のトランスポート方式をサポートしていますが、現在、安定しているのは 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 カタログの処理

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"
   

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"
   

LLM による実際の分析例：企業別のブルシット・ジョブ

「ブルシット・ジョブに就く従業員が最も多い企業を特定し、Mermaid チャートを作成して」と依頼したとき、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 が 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：LLM が主な知見を示す

LLM はデータを分析し、次のような知見を提示できます。

• ブルシットな役職に就く従業員が最も多いのは「Unknown Co」（2 人）で、ほかの企業はいずれも 1 人です
• ほとんどの企業が、ブルシット度スコアで満点の 10.0 を達成しています
• 技術系企業（BitEdge、CyberWare など）は、特に無意味な役職を生み出しているようです
• ブルシットな役職は、経営幹部または専門職の職位レベルに集中しているようです

この例は、LLM が次のことを行えることを示しています。

1. 自然言語の質問に基づいて適切な SQL クエリを生成する
2. Trino からの結果を処理して解釈する
3. データを視覚的に表現する
4. 有意義な知見と分析を提示する

API へのアクセス

Trino MCP サーバーには、データにアクセスするための API オプションが 2 つ用意されています。

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

今後の作業

現在ベータ版であり、以下の改善を予定しています。

• SSE トランスポートの問題を解決するため、利用可能になり次第、新しい MCP バージョンと統合する
• Hive、JDBC、その他のコネクターのサポートを追加／検証する
• さまざまな種類と複雑さのクエリに対する、より包括的な検証を追加する
• より多くのデータ型と高度な Trino 機能への対応を実装する
• エラー処理と復旧メカニズムを改善する
• ユーザー認証と権限制御を追加する
• より包括的な使用例とドキュメントを作成する
• 管理者向けの監視・管理インターフェースを開発する
• パフォーマンス指標とクエリ最適化のヒントを追加する
• 長時間実行されるクエリと結果のストリーミングに対応する

---

Stink Labs 開発、2025