Azure FunctionでMCPサーバーのデプロイのやり方(SQL版):その３＋ボーナス

パート1では、Azure Functionsをローカルおよびクラウドでデプロイする基本と、Model Context Protocol (MCP) インスペクターについて説明しました。パート2ではさらに一歩進み、引数を受け取るMCPツールをAzure Functionとして定義する方法を説明しました。そして今回は、パート3として、PostgreSQLデータベースとやり取りするMCPツールセットをAzure Functions上で構築する方法について深く掘り下げます。また、Azure FunctionからMCPツールがスムーズにアクセスできるようにするために、PostgreSQLをどのように設定すべきかも議論します。

これらのツールは、AIエージェントやMCPインスペクターなどの外部システムによって使用され、Azure上または互換性のあるPostgresインスタンスにホストされているPostgreSQLデータベースの構造やデータを照会し、理解するために利用されます。すべてのコードはこのGitHubリポジトリにあります。このリポジトリをクローンし、作業ディレクトリとして使用して、この記事の説明を理解してください。

---

🧠 Model Context Protocol (MCP)とは？

Model Context Protocol (MCP) は、言語モデルとバックエンドシステム間の通信を可能にする仕様です。これにより、ツールは標準化されたインターフェースを通じて機能を公開し、それらをプログラム的に発見・呼び出すことができます。
本稿では、各Azure FunctionがMCP Serverとして動作し、Azureまたは互換性のあるPostgreSQLデータベースとのやり取りに関連した特定の機能を提供します:

• データベース一覧の取得
• スキーマ情報の取得
• 読み取り専用クエリの実行
• キーおよび制約情報の取得
  各関数にはtoolName、description、toolPropertiesなどのメタデータが含まれており、MCPインスペクターがそれらを発見・利用できるようになっています。

---

🛠️ Azure Functions Python モデル v2 概要

この実装では、関数がデコレーターを使用して定義され、v1モデルよりもPythonらしいAzure Functions Python v2 プログラミングモデルを使用しています。
各ツールを表す"mcpToolTrigger"型の汎用トリガーを定義します。これらは入力パラメーターを含むコンテキスト文字列を受け取り、JSON形式のレスポンスを返します。

---

📦 ツール概要

利用可能なツールの一覧は以下の通りです:

ツール名	説明
get_databases_tool	PostgreSQLサーバー上のテンプレート以外のすべてのデータベースを取得します。
get_schemas_tool	publicスキーマ内のテーブルに対するカラムレベルのスキーマ情報を取得します。
query_data_tool	ユーザーが提供した読み取り専用SQLクエリを実行します。
get_all_keys_tool	publicスキーマ内のテーブルに対するキーおよび制約情報を取得します。

---

🔧 主なコンポーネント

1. コンテキスト引数の解析

MCPインスペクター経由で渡される動的な入力（context引数）を処理するために、_parse_context_args()を使用します:

def _parse_context_args(context_str: str, expected_arg_names: list) -> dict:
    ...

これは入力されたJSON文字列を解析し、必要なフィールドを検証して抽出された引数を返します。

2. クエリの実行

_execute_query()関数は、データベースとの実際のやり取りを担当します:

def _execute_query(query: str) -> str:
    ...

これは共有接続マネージャー(db_manager)を使用してPostgreSQLデータベースに接続し、クエリを実行し、結果をJSON形式でフォーマットします。

✅ この方法により、接続、実行、ログの一貫した処理が保証されます。

---

💡 例: get_databases_tool

この関数は、テンプレート以外のすべてのデータベースを取得します:

@app.generic_trigger(
    arg_name="context",
    type="mcpToolTrigger",
    toolName="get_databases_tool",
    description="Gets the list of all databases...",
    toolProperties="[]"
)
def get_databases_tool(context: str) -> str:
    query = "SELECT datname FROM pg_database WHERE datistemplate = false;"
    return _execute_query(query)

注: toolProperties="[]"は、入力パラメーターを必要としないことを意味しています。

---

💬 例: query_data_tool

このツールによりユーザーはカスタムのSELECTクエリを実行できます。これらのクエリはLLMによって生成されます:

_QUERY_SQL_PROPERTY = "sql_query"
tool_properties_query_data_object = [
    ToolProperty(_QUERY_SQL_PROPERTY, "string", "The SQL SELECT query to execute.")
]
tool_properties_query_data_json = json.dumps([prop.to_dict() for prop in tool_properties_query_data_object])
@app.generic_trigger(
    arg_name="context",
    type="mcpToolTrigger",
    toolName="query_data_tool",
    description="Runs read-only SQL queries...",
    toolProperties=tool_properties_query_data_json,
)
def query_data_tool(context: str) -> str:
    args = _parse_context_args(context, [_QUERY_SQL_PROPERTY])
    if "error" in args:
        return json.dumps(args)
    sql_query = args[_QUERY_SQL_PROPERTY]
    if not sql_query.strip().upper().startswith("SELECT"):
        return json.dumps({"error": "This tool is for SELECT queries only..."})
    return _execute_query(sql_query)

🔐 セキュリティ: 紛失防止のため、SELECTクエリのみ許可されています。

---

🧪 ローカルでのテストとデバッグ

ローカルでのデプロイの詳細については、パート1: MCPサーバーコードのローカルデプロイをご参照ください。以下のコマンドを実行して、これらの関数をローカルでテストできます:

source .venv/bin/activate
func start

その後、手動でエンドポイントを呼び出したり、ローカルのMCPインスペクタークライアントを使用してテストできます。MCPインスペクターの使用方法についてはパート1をご参照ください。

---

☁️ クラウドへのデプロイ

クラウドへの一般デプロイについては、パート1: Azure Function（クラウド）へのデプロイをご参照ください。Azure App ServiceがPostgreSQLデータベースにアクセスできること、およびAzureポータルで環境変数が正しく設定されていることを確認してください。PostgreSQLにConnection Stringを使用している場合は、それがAzure Functionの概要＞設定＞環境変数＞アプリケーション設定セクションに追加されていることを確認してください。

---

🔎 MCPインスペクター

これで、展開されたツールをMCPインスペクターから完全に表示できるようになりました。展開済みのURIを挿入することで利用可能です:

セル内でSQLクエリコードを書き、「Run Tool」をクリックして結果を観察できます。

📚 ボーナス: データベース接続マネージャー

Azure Functionsのようなサーバーレスアプリケーションを構築する場合、データベース接続などの外部リソースを効率よく管理することが不可欠です。以下の共有モジュールは、PostgreSQLデータベースへの接続プールを設定し、接続を取得および解放するユーティリティを提供します。これにより、状態を持たないイベント駆動型の環境であるAzure Functionsにおいても接続を効果的に再利用できるようになります。

✅ コード全体の概要

import logging
import os
import psycopg2
from psycopg2 import pool
db_pool = None
def init_db_pool():
    global db_pool
    if db_pool is None:
        try:
            conn_str = os.environ.get("POSTGRES_CONNECTION_STRING")
            if not conn_str:
                logging.error(
                    "DB_CONNECTION_STRING environment variable not set."
                )
                raise ValueError(
                    "Database connection string is not configured."
                )
            db_pool = pool.SimpleConnectionPool(1, 20, conn_str)
            logging.info("Database connection pool initialized successfully.")
        except (Exception, psycopg2.Error) as error:
            logging.error(
                f"Error while connecting to PostgreSQL or initializing pool: {error}"
            )
            db_pool = None
    return db_pool
def get_db_connection():
    """Gets a connection from the pool."""
    global db_pool
    if db_pool is None:
        init_db_pool()
        if db_pool is None:
            raise ConnectionError(
                "Database pool is not initialized. Check logs for errors."
            )
    try:
        conn = db_pool.getconn()
        if conn:
            logging.debug("Retrieved a connection from the pool.")
            return conn
        else:
            logging.error(
                "Failed to get connection from pool, pool might be exhausted or broken."
            )
            raise ConnectionError("Failed to get connection from pool.")
    except (Exception, psycopg2.Error) as error:
        logging.error(f"Error getting connection from pool: {error}")
        raise
def release_db_connection(conn):
    """Releases a connection back to the pool."""
    global db_pool
    if db_pool and conn:
        try:
            db_pool.putconn(conn)
            logging.debug("Released a connection back to the pool.")
        except (Exception, psycopg2.Error) as error:
            logging.error(f"Error releasing connection to pool: {error}")
init_db_pool()

---

📌 主なコンポーネント

1. db_pool – グローバル接続プール

単一の接続プールインスタンスが作成され、関数呼び出し全体で再利用されます。これにより、関数が毎回新しい接続を確立するオーバーヘッドを軽減できます。

2. init_db_pool() – プールの初期化

• 既にプールが存在するかチェックします。
• 存在しない場合、POSTGRES_CONNECTION_STRINGから環境変数を読み込みます。
• 最小1、最大20の接続を持つSimpleConnectionPoolを作成します。
• 成功または失敗時にログを記録します。

⚠️ 重要: 資格情報をハードコーディングしないでください。常に安全な環境変数を使用して、接続文字列などのシークレットを管理してください。

3. get_db_connection() – 接続の取得

この関数はプールから接続を取得します。プールが初期化されていることを確認したうえで、接続を取得します。取得できない場合は適切なエラーを発生させます。

4. release_db_connection(conn) – 接続の解放

データベース接続を使用した後は、この関数を使用してプールに戻す必要があります。これを怠ると、プールが枯渇する可能性があります。

---

🔁 Azure Functionsで接続プールを使う理由

Azure Functionsはステートレスで、短命であり、多くの場合並列実行されます。各呼び出しはデータベースにアクセスする必要があるかもしれません。接続プールがないと:

• 関数呼び出しごとに新しいデータベース接続を開閉します。
• 多すぎる接続が開くことでデータベースのリソースが枯渇します。
• 頻繁な接続のセットアップ/終了によりレイテンシが増加し、パフォーマンスが低下します。
  接続プールを使うことで:
• 新しい接続を作成する代わりに既存の接続を再利用できます。
• TCPハンドシェイクと認証の繰り返しによるレイテンシを削減できます。
• データベース側の接続制限に達することを防ぎます。
• 並列負荷下でのスケーラビリティと信頼性を向上させます。

💡 ヒント: Azure Functionsが複数のインスタンスにスケールアウトする可能性があるため、予想される関数インスタンス数に対応できるだけの同時接続数をデータベースが許容するようにしてください。

---

🧰 今後の拡張機能

• INSERT/UPDATE/DELETE操作をサポートする別のツールの追加
• ページングとフィルタリングの実装
• 認証と役割ベースのアクセス制御の統合
• OpenTelemetryを使ったトレーシングと監視

---

✅ 結論

Model Context Protocol (MCP) に準拠したツールをAzure Functions (Python v2モデル) で構築することで、PostgreSQLデータベースとインテリジェントなエージェントやインスペクターとのシームレスな統合が可能になります。
お読みいただきありがとうございます。このシリーズが役に立ったと思われましたら、いいねやチーム・コミュニティとの共有をお願いします。引き続き、クラウド、Python、サーバーレスに関するコンテンツを更新していきますので、どうぞお楽しみに！
🚀 Happy coding!

---

📚 リソース

• Azure Functions Python Developer Guide
• Model Context Protocol GitHub
• Model Context Protocol Introduction
• Improve database performance with connection pooling