🔌

ScalarDB Cluster 向けの MCP Server を書きました

に公開
PostgreSQL
DynamoDB
Model Context Protocol
ScalarDB
Cline
tech

はじめに

ScalarDB Cluster は強力なデータ抽象化機能を提供しています。既存の DynamoDB テーブルや PostgreSQL テーブルの上位に ScalarDB Cluster をインストールすると、下位のテーブルに対して、製品に依存しない抽象的なデータ操作レイヤーを構築できます。これはエンジニアにとって大きな負担削減になります。

最近は Cline などの AI エージェントが爆発的な進化をとげています。そのため、VSCode を使った AI 開発などで AI エージェントを活用しているエンジニアは多いのではないでしょうか。

ScalarDB Cluster は、下位のデータベースに対する操作を gRPC の SCAN, PUT, DELETE などのメソッドに変換してくれます。また、下位のデータベースの SQL 方言に依存しない共通の SQL 記法を提供しています。これ自体は便利な機能なのですが、LLM とデータを連携させるためには、エンジニアが ScalarDB Cluster からデータを取り出し、LLM にわたす処理を Python などで記述する必要があります。

この課題を解決するのが Model Context Protocol（MCP）です。MCP は、AI エージェントと外部システムを連携させるための標準化されたプロトコルで、AI エージェントが直接データベースや API などの外部リソースにアクセスできるようにします。MCP を活用することで、エンジニアが仲介コードを書かなくても、AI エージェントと ScalarDB Cluster を直接接続できるようになります。

本記事では、ScalarDB 用の Model Context Protocol（MCP）Server を実装しました。これによって、以下のメリットが得られます：

  • AI エージェントは抽象化されたデータベーススキーマを理解し、適切なクエリを生成できる
  • エンジニアは Python や SQL を書かなくても、自然言語でデータ操作を指示できる
  • データベース操作と AI の推論を組み合わせた高度なワークフローを実現できる

仕組み

ScalarDB Cluster 向けの MCP Server は、Python 製のスタンドアローンサーバーとして実装しました。このサーバーは以下のように動作します：

  1. 環境変数 SCALARDB_CLUSTER_ENDPOINT でエンドポイントを受け取る
  2. gRPC を使用して ScalarDB Cluster との接続を確立する
  3. AI エージェントからのリクエストを受け取り、適切な ScalarDB Cluster API を呼び出す
  4. 結果を AI エージェントに返す

セキュリティ面では、クエリ実行時に以下の対策を実施しています：

  • クエリ実行ツール（execute_query）では、SELECT 文のみ許可
  • セミコロンを含む複数ステートメントの禁止

ツール一覧

1. get_schema

スキーマ情報を取得するツール。特定の名前空間のテーブル構造、カラム、パーティションキー、クラスタリングキーなどの詳細情報を提供します。

2. list_namespaces

ScalarDB Cluster 内のすべての名前空間（namespace）の一覧を取得します。

3. list_tables

特定の名前空間内のすべてのテーブルの一覧を取得します。

4. execute_query

SELECT 文のみを実行できる読み取り専用クエリツール。セキュリティ上の理由から、複数のステートメントや特定のキーワードを含むクエリは実行できません。

5. execute_transaction

複数の SQL 文（INSERT、UPDATE、DELETE）をトランザクションとして実行するツール。トランザクションの開始、実行、コミット（または失敗時のロールバック）を自動的に管理します。

これらのツールを組み合わせることで、AI エージェントは ScalarDB Cluster のデータを効率的に操作し、エンジニアの質問に対して正確な回答を提供できるようになりました。

動作デモ

テーブルの探索

最初に、テーブルを探すという簡単なタスクをお願いしてみました。

質問はシンプルに、「scalardb-clusterからPDF文書のリファレンスが格納されているテーブルを探して下さい。」としました。ここでは探す場所を具体的に指定していますが、これはローカルフォルダを検索する時間を節約するためであり、必須ではありません。ScalarDB Cluster の下位にあるデータベースには、RAG システムを構成する複数のテーブルが存在している状態です。

Cline がタスクを始めました。（ログの一部を読みやすいように省略しています）

まず、scalardb-clusterの名前空間（namespace）のリストを取得して、どのような名前空間があるかを確認しましょう。そのためには、list_namespacesツールを使用します。

list_namespaces:
Arguments
なし

名前空間のリストが取得できました。2つの名前空間があります：

rag_demo
scalardb

これらの名前空間の中から、PDF文書のリファレンスが格納されているテーブルを探す必要があります。「rag_demo」という名前空間は「RAG（Retrieval-Augmented Generation）」に関連している可能性があり、文書検索に関連する可能性が高いです。

最初に使ってほしいツールを認識してくれました。そして、名前空間の rag という文字列から、「ここに PDF の情報がありそう」という推測をしてくれました。

「rag_demo」名前空間内のテーブルリストを取得します。

list_tables:
Arguments
{
  "namespace": "rag_demo",
}

テーブルのリストが取得できました。3つのテーブルがあります：

reference
sentence
summary

これらのテーブルの中で、「reference」というテーブルがPDF文書のリファレンスを格納している可能性が高いです。テーブル名から推測すると、ドキュメントの参照情報を保存するテーブルと思われます。

今度はテーブル名の reference と、質問文の リファレンス の一致に注目しました。質問でヒントを与えすぎているかもしれませんが、正しいです。

「rag_demo」名前空間内のスキーマ情報を取得します。

get_schema:
Arguments
{
  "namespace": "rag_demo",
}

「rag_demo」名前空間のスキーマ情報が取得できました。特に「reference」テーブルに注目すると、以下の列があります：

reference_id (TEXT型) - パーティションキー
file_name (TEXT型)
file_link (TEXT型)

このテーブルの構造から、PDF文書のリファレンス情報を格納していると考えられます。特に以下の点から判断できます：

reference_id: 各リファレンスの一意の識別子
file_name: ファイル名を格納
file_link: おそらくPDFファイルへのリンクを格納

テーブルのスキーマを確認し、データの意味を推測していますね。

実際のデータを確認するために、「reference」テーブルからいくつかのレコードを取得します。

execute_query: 
Arguments
{
  "namespace": "rag_demo",
  "query": "SELECT * FROM reference LIMIT 5"
}

「reference」テーブルからデータを取得しました。3つのレコードが返されました。これらのレコードには以下の情報が含まれています：

レコード1:
reference_id: AAAA-AAAA
file_name: 文書A.pdf
file_link: https://fileserver/pdf/A.pdf

レコード2:
reference_id: BBBB-BBBB
file_name: 文書B.pdf
file_link: https://fileserver/pdf/B.pdf

レコード2:
reference_id: CCCC-CCCC
file_name: 文書C.pdf
file_link: null

これらのデータから、「reference」テーブルはPDF文書のリファレンス情報を格納していることが確認できました。file_nameはPDFファイルの名前を示し、file_linkはPDFファイルへのリンクを示しています（ただし、一部のレコードではnullになっています）。

実際のレコードの値を確認しています。LIMIT 句をつけているところや、null を気にしているあたりに人間味がありますね。

また、「sentence」テーブルも「reference_id」をパーティションキーとして持っているため、「reference」テーブルと関連していると考えられます。「sentence」テーブルは各PDFの文章を格納している可能性があります。

他に、外部キーとして参照しているテーブルを特定し、テーブルの役割を推察していますね。

これでタスクは完了しました。問題解決のアプローチも含めて、探索結果に間違いはありませんでした。

動作デモ「テーブルの探索」におけるシーケンス図を以下に示します。

データの更新

次に、データのメンテナンスをお願いしてみましょう。

質問は、「リファレンステーブルから、ファイルリンクがNULLのレコードを消して下さい。」としました。

Cline がタスクを開始します。

前回の作業で、scalardb-clusterの「rag_demo」名前空間内の「reference」テーブルを調査し、このテーブルにはPDF文書のリファレンス情報が格納されていることを確認しました。また、いくつかのレコードでは「file_link」列がNULLになっていることも確認しました。

今回の依頼を達成するためには、以下のステップを実行する必要があります：

「reference」テーブルから「file_link」がNULLのレコードを削除するSQLステートメントを作成する
そのSQLステートメントをトランザクション内で実行する

計画をたてていますが、正しいですね。

まず、「reference」テーブルから「file_link」がNULLのレコードがいくつあるかを確認しましょう。

execute_query: 
Arguments
{
  "namespace": "rag_demo",
  "query": "SELECT COUNT(*) FROM reference WHERE file_link IS NULL"
}

削除対象のレコードの件数を確認しようとしていますが、この SQL はエラーになってしまいます。

エラーが発生しました。エラーメッセージは以下の通りです：

DB-SQL-10026: Syntax error. Line 1:12 no viable alternative at input 'SELECT COUNT...

COUNT関数の使用でエラーが発生しました。ScalarDBのSQLクエリでは一部の標準SQL機能がサポートされていないようです。

Cline がエラーメッセージから読み取ったとおり、ScalarDB Cluster が提供する抽象化された SQL の記法では、COUNT(*) はサポートされていません。これは、分散データベースにとって全列を指定してレコード数をカウントすることは非常にコストが高い操作になるためです。

こういった場合でも、Cline はワークアラウンドを考えてタスクを継続してくれます。

代わりに、「reference」テーブルからすべてのレコードを取得して、file_linkがNULLのレコードを確認します。

execute_query: 
Arguments
{
  "namespace": "rag_demo",
  "query": "SELECT * FROM reference"
}

「reference」テーブルからすべてのレコードを取得しました。全部で3つのレコードがあり、そのうち1つのレコードでfile_linkがNULLになっています：

全レコードを取得して値をチェックするという強引なワークアラウンドを行っていますね。これは結局コストが高いオペレーションなので、実運用では追加のプロンプトや Cline の共通設定でこういった制約事項をインプットしてあげる必要があるでしょう。ここでは、Cline がタスクを完遂しようとしてくれている姿勢を評価してあげるべきです。

Cline は削除処理に進んでいきます。

削除するレコードのreference_idは以下の1つです：

CCCC-CCCC

ユーザーの依頼に従って、file_linkがNULLの2つのレコードを削除します。ScalarDBでは、トランザクション内でDELETE文を実行する必要があるため、execute_transactionツールを使用します。

execute_transaction: 
Arguments
{
  "namespace": "rag_demo",
  "statements": [
    "DELETE FROM reference WHERE reference_id = 'CCCC-CCCC'",
  ]
}

正しいステートメントを構築できていますね。

execute_transaction の Auto-approve を無効にしておけば、ここで Cline のタスクが中断され、ユーザーに操作の承認を求めてきます。

ここでユーザーが操作を承認すればタスクが再開され、データが削除されます。今回のデモは実際にレコードは消さずに、承認を Reject して終わります。

こちらもタスクの実行内容について、間違っている部分はありませんでした。

動作デモ「データの更新」におけるシーケンス図を以下に示します。

その他にできること

本記事では省略しますが、質問に「外部結合している他のテーブルのレコードも消して下さい。」といったフレーズを追加することで、関連テーブルも含めた削除ステートメントを生成し、それらをまとめて1つのトランザクションで実行するといった高度なことも行えます。

また、DDL 操作を行うツールを追加して、AI エージェントによって作成したデータを新規テーブルとして保存したり、既存のテーブル構造のリファクタリングタスクを行ってもらうことも可能になるでしょう。

まとめ

本記事では、ScalarDB Cluster 用の Model Context Protocol（MCP）Server の実装について解説しました。

ScalarDB Cluster のデータの抽象化機能は、AI エージェントによるデータベース操作と相性が良いように思いました。

ソースコード

ソースコードは以下のリポジトリで公開しています。
https://github.com/naomine-biz/mcp-server-scalardb

Discussion