目的
MCP(Model Context Protocol)に関する記事を多く見かけるようになりました。
弊社ではDB2をデータベースとして用いることが多いため、MCPを使って、DB2データベースにアクセスするMCPサーバーを作成する方法をまとめました。
今回はClaude Desktopからローカル環境のDB2へアクセスする方法をご紹介します。SELECT文のみを実行できる読み取り専用のツールになっています。
対象読者
・DB2をデータベースとして使用している方
・DB2の導入を検討されている方や生成AIの活用を検討されている方
・MCPサーバーの構築に興味のある方
参考文献
今回の記事を作成するにあたり、参考にさせて頂いた記事を列挙します。
環境設定について
Python 3.13以下のバージョン
→今回使用するIBM_DBが3.13までしかサポートしていないため
DB2 Databaseへのアクセス権限
→IDとパスワードが分かる状態です。
uv（高速なPythonパッケージマネージャー）
→必要に応じてインストールをお願いします。
必要な環境変数の設定（.envファイル）
→本記事内にて作成します。
目次
1章 MCPとは
公式ドキュメントには下記の様な説明があります。
MCPは、アプリケーションがLLMにコンテキストを提供する方法を標準化するオープンプロトコルです。MCPはAIアプリケーション用のUSB-Cポートのようなものだと考えてください。USB-Cがデバイスを様々な周辺機器やアクセサリに接続するための標準化された方法を提供するのと同様に、MCPはAIモデルを様々なデータソースやツールに接続するための標準化された方法を提供します。
かくいう私もLLMと外部リソース(データベースやAPI等)の連携を簡単にするための仕組み位の認識でして、今後も知識を吸収していかねばと思っています。
上記の公式ドキュメントにはMCPサーバー構築のガイダンスもあり、本記事を作成するにあたり参考にしました。
MCP自体についてはこちらの資料が一番イメージを付けやすかったので、ご紹介します。
今回のケースであれば
MCPクライアント: Claude Desktop
MCPサーバー: 仮想環境上にpythonスクリプトを用いて構築
ツール: SELECT文を用いた読み取り専用のツール
リソース: DB2データベース
になるイメージだと思います。
次の章ではMCPサーバーの構築方法をご紹介します。
2章 DB2用MCP-Server作成方法について
任意のdb2-mcp-server用のフォルダを作成します。
ここではC:\db2-mcp-serverとしています。
次にDB2にアクセスするための.envファイルを作成します。
USER、PASSWORD、DATABASEには各環境の値を設定して下さい。
HOST、PORTについてはローカル環境にDB2データベースがある場合を想定しています。
DB_USER=*******
DB_PASSWORD=********
DB_DATABASE=*******
DB_HOST=localhost
DB_PORT=50000
次にrequirements.txtを作成します。
最新のibm_dbをインストールするとdllファイルのloadエラーが発生するため、ここではマイナーバージョンを指定しています。
ibm_db==3.1.4
python-dotenv
sqlparse
mcp
DB2に対してSELECT文を投げて、結果をJSONで返すシンプルなMCPサーバーを実装するためのPythonファイル(mcp_server_db2.py)を作成します。
Claude DesktopなどのMCPクライアントから呼び出すことで、対話的にDB2のデータを取得・可視化できます。
ここではSELECT文しか出来ませんので、必要に応じてスクリプトを修正してください。
コード全体は付録になります。
①モジュールの導入
ibm_db_dbi：DB2公式のPython用ドライバーです。
dotenv：.envファイルから環境変数を読み込むために使用します。
fastmcp：ClaudeなどのMCPクライアントとの通信を行うライブラリです。
import ibm_db_dbi
import os
from dotenv import load_dotenv
import json
import atexit
from mcp.server.fastmcp import FastMCP
import mcp.types as types
# グローバル変数
conn = None
cursor = None
②DB接続の初期化と修了処理の設定
.envに記載されたユーザー名やパスワード、ホスト名などから接続文字列を構築します。
atexit.register()を使って、スクリプト終了時に接続を安全にクローズするようにしています。
def initialize_db_connection():
global conn, cursor
load_dotenv()
user = os.getenv("DB_USER")
password = os.getenv("DB_PASSWORD")
database = os.getenv("DB_DATABASE")
host = os.getenv("DB_HOST")
port = os.getenv("DB_PORT")
conn_str = (
f"DATABASE={database};HOSTNAME={host};PORT={port};"
f"PROTOCOL=TCPIP;UID={user};PWD={password};"
)
conn = ibm_db_dbi.connect(conn_str, "", "")
cursor = conn.cursor()
def cleanup_db_connection():
global conn, cursor
if cursor:
cursor.close()
if conn:
conn.close()
atexit.register(cleanup_db_connection)
③SELECT専用のクエリ実行関数の作成
SELECT文以外は拒否するようになっています。
結果はPythonの辞書型に整形され、json.dumps()で整ったJSONに変換されます。
ClaudeなどのMCPクライアントでも読み取りやすくなっています。
def execute(query: str) -> str:
global cursor
query = query.strip().rstrip(";")
if not query.lower().startswith("select"):
raise ValueError("このツールはSELECT文のみを実行できます")
cursor.execute(query)
rows = cursor.fetchall()
columns = [desc[0] for desc in cursor.description]
results = [dict(zip(columns, row)) for row in rows]
return json.dumps(results, ensure_ascii=False, indent=2)
mcp = FastMCP("DB2")
④MCPツールの定義設定
この関数がMCPクライアントから呼び出されるエンドポイントとなります。
成功時は結果、失敗時は例外の内容を文字列で返します。
@mcp.tool(
name="execute_db2",
description="DB2に対してSELECT文を実行します。"
)
def execute_db2(query: str) -> str:
try:
result = execute(query)
return [types.TextContent(type="text", text=result)]
except Exception as e:
return [types.TextContent(type="text", text=f"エラー: {str(e)}")]
⑤サーバーの起動処理設定
スクリプトが直接実行された場合、DB接続を初期化した上でMCPサーバーが起動します。
通信方式には標準入出力（stdio）を採用しており、Claude DesktopなどのローカルMCPクライアントとシームレスに通信可能です。
if __name__ == "__main__":
initialize_db_connection()
try:
mcp.run(transport="stdio")
finally:
cleanup_db_connection()
以上がdb2データベースにSELECT文を実行するmcp_serverを構築するためのPythonスクリプトになります。
作成した3ファイルを冒頭で作成したフォルダに保存してください。
フォルダの場所でコマンドプロンプトを開いて仮想環境を作成します。
ここでは検証にあたり、問題なく作成できたPython 3.10を指定して作成しています。
py -3.10 -m venv .venv
.venv\Scripts\activate
仮想環境でパッケージをインストールします。
uv pip install -r requirements.txt
ここで下記のコマンドを実行すると参照エラーが起きると思います。
python
import ibm_db
FileNotFoundError: [WinError 3] 指定されたパスが見つかりません。
: 'C:\\db2-mcp-server\\.venv\\lib\\site-packages\\clidriver\\bin'
上記ページにありますが、最新のibm_dbではdll loadエラーが起き、過去のバージョンでもcridriverフォルダの参照エラーが起きます。
後者については下記の様な形にすると参照エラーを回避できますが、都度手動実行が必要になります。
Claude Desktop起動時に、db2_mcp_serverも自動起動する必要があるため、今回のケースでは適しません。
また、環境変数に設定しても解決しませんでした。
import os
os.add_dll_directory('path to clidriver installation until bin')
import ibm_db
ibm_db.pyを修正する等の解決策もあるとは思うのですが、
今回はclidriverフォルダを'C:\db2-mcp-server\.venv\lib\site-packages\に配置することで回避したいと思います。
clidriverに関しては下記ページよりダウンロードできますので、環境に合ったものをインストールしていただき、配置をお願いします。
＊＊お願い＊＊
このibm_dbの参照エラーと最新バージョンのdllファイルのロードエラーに関しては手探り状態の回避策を記載しております。
もし、他の問題が起きた方や他の解決策をお持ちの方はぜひコメント欄にて教えて頂けますと幸いです。
ここまでがDB2用のMCPサーバーの構築方法になります。
3章 MCPクライアントの導入について
今回はMCPクライアントとしてClaude Desktopを使用します。
下記からダウンロードできます。.exeファイルを実行することで簡単に使用できます。
次に、Claude Desktop側のセットアップを行います。
起動したらヘルプ>開発者モードを有効にするをクリックします。
ファイルタブから設定をクリックしてください。
開発者をクリックして、構成を編集をクリックします。
エクスプローラーが開きますので、その中の「claude_desktop_config.json」を編集します。
内容は下記になります。
{
"mcpServers": {
"DB2": {
"command": "C:\db2-mcp-server\.venv\Scripts\python.exe",
"args": [
"C:\db2-mcp-server\mcp_server_db2.py"
]
}
}
}
セットアップは以上になります。端末を再起動してください。
Claude Desktopを起動して、開発者タブで先ほど設定したDB2が「running」になっていることを確認してください。
ここまでで準備は完了しましたので、実際に使ってみたいと思います。
まず対話画面で＋ボタンの横のボタンをクリックすると、先ほど設定したDB2が追加されています(上)のでクリックして、選択状態にしてください。(下)
最初の実行時には「〇〇からのツールを許可しますか？」というウィンドウが出る場合もありますので、「このチャットで許可する」か「一度だけ許可」をクリックします。
下記が使用した時の例になります。
今回はDB2内にテスト用の架空の札幌営業所の売上データを配置しています。
テーブルの内容についてもまとめてくれます。
テーブルの内容を基にグラフを作成することも出来ます。
あとがき
今回はDB2用の簡単なMCPサーバー構築方法をご紹介しました。
対話式でリソースからグラフを作成することができ、データの可視化のハードルが下がった印象を受けます。
今回はSELECT文を使った読み取り専用のツールを作成しましたが、より複雑な処理をツールに組み込むことで様々なケースにも対応できると思いますので、改善していきたいと思っています。
記事の内容につきましてご不明点、ご指摘ございましたらお気軽にコメントください。
拙い文章ではございますが、最後までお付き合いいただきありがとうございました。
付録
import ibm_db_dbi
import os
from dotenv import load_dotenv
import json
import atexit
from mcp.server.fastmcp import FastMCP
import mcp.types as types
# グローバル変数
conn = None
cursor = None
def initialize_db_connection():
global conn, cursor
load_dotenv()
user = os.getenv("DB_USER")
password = os.getenv("DB_PASSWORD")
database = os.getenv("DB_DATABASE")
host = os.getenv("DB_HOST")
port = os.getenv("DB_PORT")
conn_str = (
f"DATABASE={database};HOSTNAME={host};PORT={port};"
f"PROTOCOL=TCPIP;UID={user};PWD={password};"
)
conn = ibm_db_dbi.connect(conn_str, "", "")
cursor = conn.cursor()
def cleanup_db_connection():
global conn, cursor
if cursor:
cursor.close()
if conn:
conn.close()
atexit.register(cleanup_db_connection)
def execute(query: str) -> str:
global cursor
query = query.strip().rstrip(";")
if not query.lower().startswith("select"):
raise ValueError("このツールはSELECT文のみを実行できます")
cursor.execute(query)
rows = cursor.fetchall()
columns = [desc[0] for desc in cursor.description]
results = [dict(zip(columns, row)) for row in rows]
return json.dumps(results, ensure_ascii=False, indent=2)
mcp = FastMCP("DB2")
@mcp.tool(
name="execute_db2",
description="DB2に対してSELECT文を実行します。"
)
def execute_db2(query: str) -> str:
try:
result = execute(query)
return [types.TextContent(type="text", text=result)]
except Exception as e:
return [types.TextContent(type="text", text=f"エラー: {str(e)}")]
if __name__ == "__main__":
initialize_db_connection()
try:
mcp.run(transport="stdio")
finally:
cleanup_db_connection()