MCP SQL サーバー

会話型 AI インターフェースを介して SQL データベース対話ツールを提供する FastMCP サーバー。

概要

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

特徴

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

要件

Python 3.7以上
pyodbc
非同期
FastMCPフレームワーク
マイクロソフトSQLサーバー
SQL Server 用 ODBC ドライバー 17

インストール

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

pip install pyodbc asyncio fastmcp

Microsoft SQL Server と SQL Server 用の ODBC ドライバー 17 がインストールされていることを確認してください。
スクリプトで接続設定を構成します。

# 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 データベースへの接続を確立します。

利用可能なツール

クエリSQL

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

query_sql(query: str = None) -> str

クエリが指定されていない場合は、デフォルトでSELECT * FROM [dbo].[Table_1]になります。
クエリ結果をフォーマットされた文字列として返します

リストテーブル

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

list_tables() -> str

テーブル名のリストを文字列として返します

テーブルの説明

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

describe_table(table_name: str) -> str

table_name : 記述するテーブルの名前
名前やデータ型などの列情報を返します

非クエリ実行

INSERT、UPDATE、DELETE などの非クエリ SQL ステートメントを実行します。

execute_nonquery(sql: str) -> str

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

リスト_odbc_drivers

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

list_odbc_drivers() -> str

インストールされているODBCドライバのコンマ区切りリストを返します。

データベース情報

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

database_info() -> str

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

建築

サーバーは、操作のブロックを回避するために非同期アーキテクチャを使用します。

ライフサイクル管理: app_lifespanコンテキストマネージャーは、データベース接続のセットアップと破棄を処理します。
非ブロッキング操作: データベース操作は、メインイベントループのブロックを防ぐためにasyncio.get_event_loop().run_in_executor()を使用して別のスレッドで実行されます。
エラー処理: すべての操作には、便利なエラーメッセージによる包括的なエラー処理が含まれます。

エラー処理

サーバーはさまざまなエラー状態を処理します。

データベース接続の失敗
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クエリの入力検証を実装することを検討してください。
最小権限の原則に基づいてデータベースユーザーの権限を制限する

トラブルシューティング

よくある問題:

接続エラー: SQL Server インスタンス名を確認し、実行中であることを確認します
ODBC ドライバーエラー: SQL Server 用 ODBC ドライバー 17 がインストールされていることを確認してください
権限エラー: アプリケーションを実行している Windows ユーザーが適切な SQL Server 権限を持っていることを確認してください

ライセンス

[ライセンス情報]

接触

[連絡先]

MCP SQL Server

Integrations