MCP SQL Server

MCP SQL Server

会話型 AI インターフェースを通じて SQL データベースを操作するツールを提供する FastMCP サーバーです。

概要

このプロジェクトでは、会話型 AI インターフェースを通じて MS SQL Server の操作機能を公開するサーバーを作成します。FastMCP フレームワークを使用して SQL データのクエリや操作を行うツールを提供し、ユーザーが自然言語でデータベースを操作できるようにします。

機能

• SQL クエリを実行し、結果を表示
• データベースで利用可能なテーブルを一覧表示
• カラム情報を含むテーブル構造を表示
• クエリ以外の操作（INSERT、UPDATE、DELETE）を実行
• システムで利用可能な ODBC ドライバーを一覧表示
• データベース情報とサーバーの詳細を表示

要件

• Python 3.7+
• pyodbc
• asyncio
• FastMCP フレームワーク
• Microsoft SQL Server
• ODBC Driver 17 for SQL Server

インストール

1. Python の依存関係をインストールします。

pip install pyodbc asyncio fastmcp

2. Microsoft SQL Server と ODBC Driver 17 for SQL Server がインストールされていることを確認します。

3. スクリプト内の接続設定を構成します。

# Connection parameters
SERVER = "server\\instance"  # Change to your SQL Server instance
DATABASE = "db_name"              # Change to your database name

使用方法

サーバーを起動します。

python mcp_sql_server.py

サーバーが初期化され、指定した SQL Server データベースへの接続が確立されます。

利用可能なツール

query_sql

SQL クエリを実行し、結果を返します。

query_sql(query: str = None) -> str

• クエリが指定されていない場合、デフォルトで SELECT * FROM [dbo].[Table_1] が使用されます
• クエリ結果を整形された文字列として返します

list_tables

データベースで利用可能なすべてのテーブルを一覧表示します。

list_tables() -> str

• テーブル名の一覧を文字列として返します

describe_table

指定したテーブルの構造を取得します。

describe_table(table_name: str) -> str

• table_name：構造を取得するテーブルの名前
• 名前やデータ型などのカラム情報を返します

execute_nonquery

INSERT、UPDATE、DELETE、その他のクエリ以外の SQL 文を実行します。

execute_nonquery(sql: str) -> str

• sql：実行する SQL 文
• 影響を受けた行数を含む操作結果を返します
• トランザクションを自動的に処理します（コミット／ロールバック）

list_odbc_drivers

システムで利用可能なすべての ODBC ドライバーを一覧表示します。

list_odbc_drivers() -> str

• インストール済みの ODBC ドライバーをカンマ区切りで返します

database_info

接続中のデータベースに関する一般情報を取得します。

database_info() -> str

• サーバー名、データベース名、SQL Server のバージョン、現在のサーバー時刻、テーブル数を返します

アーキテクチャ

このサーバーでは、ブロッキング処理を避けるために非同期アーキテクチャを使用しています。

1. ライフサイクル管理：app_lifespan コンテキストマネージャーが、データベース接続の確立と終了を処理します。

2. ノンブロッキング処理：データベース操作は asyncio.get_event_loop().run_in_executor() を使用して別スレッドで実行され、メインのイベントループがブロックされるのを防ぎます。

3. エラー処理：すべての操作に包括的なエラー処理が含まれており、有用なエラーメッセージを提供します。

エラー処理

このサーバーはさまざまなエラー状況に対応します。

• データベース接続の失敗
• SQL クエリの構文エラー
• テーブルが見つからないエラー
• 権限に関する問題

すべてのエラーはログに記録され、適切なエラーメッセージがクライアントに返されます。

カスタマイズ

新しいデータベースツールを追加したり、既存のツールを変更したりするには、既存のツールで使用されているパターンに従います。

@mcp.tool()
async def your_new_tool(ctx: Context, param1: str) -> str:
    """Documentation for your tool"""
    try:
        conn = ctx.request_context.lifespan_context["conn"]
        
        if conn is None:
            return "Database connection is not available."
            
        def your_db_operation():
            # Your database operations here
            pass
            
        loop = asyncio.get_event_loop()
        result = await loop.run_in_executor(None, your_db_operation)
        
        # Process and return results
        return "Your result"
    except Exception as e:
        return f"Error: {str(e)}"

セキュリティに関する考慮事項

• このサーバーは Windows 認証（"Trusted_Connection=yes"）を使用します
• SQL インジェクションを防ぐため、SQL クエリの入力検証を実装することを検討してください
• 最小権限の原則に基づいてデータベースユーザーの権限を制限してください

トラブルシューティング

よくある問題：

1. 接続エラー：SQL Server インスタンス名を確認し、実行中であることを確認してください
2. ODBC ドライバーのエラー：ODBC Driver 17 for SQL Server がインストールされていることを確認してください
3. 権限エラー：アプリケーションを実行している Windows ユーザーに適切な SQL Server 権限があることを確認してください

ライセンス

[ライセンス情報]

お問い合わせ

[連絡先情報]