手を動かして学ぶ！MCPステップバイステップ実践ガイド for Beginners – Vol.8 秘密の合言葉！APIキーでMCPアクセスに認証を追加する #Python

# client.py (Vol.8 版)
import requests
import json
from requests.exceptions import JSONDecodeError # Vol.6 で既にインポート済み

BASE_SERVER_URL = "http://127.0.0.1:5000/devices"
# ★★★ 新しい変更点：クライアントが使用するAPIキー ★★★
# このキーはサーバーの VALID_API_KEY と一致させる必要があります。
API_KEY = "MySuperSecretMCPApiKey_Vol8" 

# display_device_info関数 (Vol.6 と同じなので省略)
def display_device_info(device_info):
    """デバイス情報を整形して表示するヘルパー関数"""
    print(f"  モデル名: {device_info.get('modelName', '情報なし')}")
    print(f"  デバイスID: {device_info.get('deviceId', '情報なし')}")
    print(f"  設置場所: {device_info.get('location', '情報なし')}")

    status_info = device_info.get('status', {})
    if status_info:
        print("  ステータス:")
        for key, value in status_info.items():
            if isinstance(value, bool):
                print(f"    {key.replace('is', '').capitalize() if 'is' in key else key.capitalize()}: {'オン' if value else 'オフ'}")
            else:
                print(f"    {key.capitalize()}: {value} {device_info.get('status', {}).get('unit', '') if key == 'currentTemperature' or key == 'targetTemperature' or key == 'brightness' else ''}")
    else:
        print("  ステータス: 情報なし")

    if "supportedModes" in device_info and device_info["supportedModes"]:
        print(f"  対応モード: {', '.join(device_info['supportedModes'])}")
    if "supportedEvents" in device_info and device_info["supportedEvents"]:
        print(f"  対応イベント: {', '.join(device_info['supportedEvents'])}")
    print("-" * 30)


def get_all_device_infos():
    target_url = BASE_SERVER_URL
    print(f"\n--- 全てのMCPデバイス情報を取得します ---")
    print(f"アクセスするURL: {target_url}")
    
    # ★★★ 変更点：リクエストヘッダーにAPIキーを追加 ★★★
    headers = {"X-API-Key": API_KEY}

    try:
        # ★★★ 変更点：headers パラメータを追加 ★★★
        response = requests.get(target_url, headers=headers, timeout=5)
        # エラー処理部分はVol.6 と同様 (HTTPError 401 がここで検知されるはず)
        response.raise_for_status() 

        data = response.json()
        # ... (以降の成功時処理はVol.6 と同じ)
        devices = data.get("devices", [])
        if devices:
            print(f"--- 取得成功 (全 {len(devices)} デバイス) ---")
            for device in devices:
                display_device_info(device)
        else:
            print("デバイス情報が見つかりませんでした。")

    except requests.exceptions.HTTPError as e: # 401 Unauthorized もここでキャッチされる
        print(f"エラー: HTTPエラーが発生しました。ステータスコード: {e.response.status_code}")
        try:
            error_details = e.response.json() # サーバーからのJSONエラーメッセージを期待
            print(f"  サーバーからのメッセージ: {error_details.get('message', json.dumps(error_details))}")
        except JSONDecodeError:
            print(f"  サーバーからの応答はJSON形式ではありませんでした。内容(一部): {e.response.text[:200]}...")
    # ... (ConnectionError, Timeout, JSONDecodeError, RequestException のキャッチはVol.6 と同じ)
    except requests.exceptions.ConnectionError:
        print(f"エラー: サーバー ({target_url}) への接続に失敗しました。")
    except requests.exceptions.Timeout:
        print(f"エラー: サーバー ({target_url}) からの応答がタイムアウトしました。")
    except JSONDecodeError: # response.json() で失敗した場合 (raise_for_statusを通った後)
        print(f"エラー: サーバーからの正常応答が正しいJSON形式ではありません。")
        if 'response' in locals() and response is not None:
             print(f"  応答内容(先頭200文字): {response.text[:200]}...")
    except requests.exceptions.RequestException as e:
        print(f"エラー: リクエスト中に予期せぬ問題が発生しました ({target_url}): {e}")


def get_specific_device_info(device_id):
    target_url = f"{BASE_SERVER_URL}/{device_id}"
    print(f"\n--- MCPデバイスID '{device_id}' の情報を取得します ---")
    print(f"アクセスするURL: {target_url}")

    # ★★★ 変更点：リクエストヘッダーにAPIキーを追加 ★★★
    headers = {"X-API-Key": API_KEY}

    try:
        # ★★★ 変更点：headers パラメータを追加 ★★★
        response = requests.get(target_url, headers=headers, timeout=5)

        # エラー処理部分はVol.6 と同様、ただし401のケースも考慮
        if response.status_code == 200:
            device_data = response.json()
            print("--- 取得成功 ---")
            display_device_info(device_data)
        elif response.status_code == 401: # Unauthorized
            error_data = response.json()
            print("--- 認証失敗 (401 Unauthorized) ---")
            print(f"  サーバーからのメッセージ: {error_data.get('message', 'APIキーが無効か、提供されていません。')}")
        elif response.status_code == 404: # Not Found
            # ... (Vol.6 と同じ)
            try:
                error_data = response.json()
                print("--- 取得失敗 (404 Not Found) ---")
                print(f"  エラーメッセージ: {error_data.get('error', 'サーバーからのエラー詳細なし')}")
                print(f"  リクエストしたID: {error_data.get('requested_id', device_id)}")
            except JSONDecodeError:
                print("--- 取得失敗 (404 Not Found) ---")
                print(f"  サーバーからのエラーメッセージはJSON形式ではありませんでした。応答: {response.text[:200]}...")
        else: # その他のHTTPエラー
            print(f"--- エラー (ステータスコード: {response.status_code}) ---")
            try:
                error_details = response.json()
                print("  サーバーからの詳細:")
                print(json.dumps(error_details, indent=2, ensure_ascii=False))
            except JSONDecodeError:
                print(f"  サーバーからの応答はJSON形式ではありませんでした。内容: {response.text[:200]}...")
    # ... (ConnectionError, Timeout, JSONDecodeError, RequestException のキャッチはVol.6 と同じ)
    except requests.exceptions.ConnectionError:
        print(f"エラー: サーバー ({target_url}) への接続に失敗しました。サーバーが起動しているか確認してください。")
    except requests.exceptions.Timeout:
        print(f"エラー: サーバー ({target_url}) への接続がタイムアウトしました。")
    except JSONDecodeError: # response.json() で失敗した場合
        print(f"エラー: サーバーからの応答 ({target_url}) が正しいJSON形式ではありません。")
        if 'response' in locals() and response is not None:
             print(f"  応答内容(先頭200文字): {response.text[:200]}...")
    except requests.exceptions.RequestException as e:
        print(f"エラー: リクエスト中に問題が発生しました ({target_url}): {e}")


def add_new_device(device_data):
    target_url = BASE_SERVER_URL
    print(f"\n--- 新しいデバイスを登録します ---")
    # ... (送信データ表示はVol.6 と同じ)
    print(f"送信先URL: {target_url}")
    print(f"送信データ:")
    print(json.dumps(device_data, indent=2, ensure_ascii=False))
    
    # ★★★ 変更点：リクエストヘッダーにAPIキーを追加 ★★★
    headers = {"X-API-Key": API_KEY}

    try:
        # ★★★ 変更点：headers パラメータを追加 ★★★
        response = requests.post(target_url, json=device_data, headers=headers, timeout=5)

        # エラー処理部分はVol.6 と同様、ただし401のケースも考慮
        if response.status_code == 201: # Created
            # ... (成功時処理はVol.6 と同じ)
            print("--- 登録成功 (201 Created) ---")
            new_device = response.json()
            display_device_info(new_device)
        elif response.status_code == 401: # Unauthorized
            error_data = response.json()
            print("--- 認証失敗 (401 Unauthorized) ---")
            print(f"  サーバーからのメッセージ: {error_data.get('message', 'APIキーが無効か、提供されていません。')}")
        # ... (409, 400 やその他のエラー処理はVol.6 と同じ)
        elif response.status_code == 409: 
            error_data = response.json()
            print("--- 登録失敗 (409 Conflict) ---")
            print(f"  エラーメッセージ: {error_data.get('error', 'サーバーからのエラー詳細なし')}")
            print(f"  競合したID: {error_data.get('deviceId', '情報なし')}")
        elif response.status_code == 400: 
            error_data = response.json()
            print("--- 登録失敗 (400 Bad Request) ---")
            print(f"  エラーメッセージ: {error_data.get('error', 'サーバーからのエラー詳細なし')}")
        else:
            print(f"--- エラー (ステータスコード: {response.status_code}) ---")
            try:
                error_details = response.json()
                print("  サーバーからの詳細:")
                print(json.dumps(error_details, indent=2, ensure_ascii=False))
            except JSONDecodeError:
                print(f"  サーバーからの応答はJSON形式ではありませんでした。内容: {response.text[:200]}...")
    # ... (ConnectionError, Timeout, RequestException のキャッチはVol.6 と同じ)
    except requests.exceptions.ConnectionError:
        print(f"エラー: サーバー ({target_url}) への接続に失敗しました。")
    except requests.exceptions.Timeout:
        print(f"エラー: サーバー ({target_url}) への接続がタイムアウトしました。")
    except requests.exceptions.RequestException as e:
        print(f"エラー: リクエスト中に問題が発生しました ({target_url}): {e}")


def update_device_info(device_id, updated_data):
    target_url = f"{BASE_SERVER_URL}/{device_id}"
    print(f"\n--- デバイスID '{device_id}' の情報を更新します ---")
    # ... (送信データ表示はVol.6 と同じ)
    print(f"送信先URL: {target_url}")
    print(f"送信データ:")
    print(json.dumps(updated_data, indent=2, ensure_ascii=False))

    # ★★★ 変更点：リクエストヘッダーにAPIキーを追加 ★★★
    headers = {"X-API-Key": API_KEY}

    try:
        # ★★★ 変更点：headers パラメータを追加 ★★★
        response = requests.put(target_url, json=updated_data, headers=headers, timeout=5)
        
        # エラー処理部分はVol.6 と同様、ただし401のケースも考慮
        if response.status_code == 200: # OK
            # ... (成功時処理はVol.6 と同じ)
            print("--- 更新成功 (200 OK) ---")
            device_info = response.json()
            if "warning" in device_info: # Vol.6 の警告メッセージ処理
                print(f"  サーバーからの警告: {device_info['warning']}")
                display_device_info(device_info.get("updated_device", {}))
            else:
                display_device_info(device_info)
        elif response.status_code == 401: # Unauthorized
            error_data = response.json()
            print("--- 認証失敗 (401 Unauthorized) ---")
            print(f"  サーバーからのメッセージ: {error_data.get('message', 'APIキーが無効か、提供されていません。')}")
        # ... (404, 400 やその他のエラー処理はVol.6 と同じ)
        elif response.status_code == 404: 
            error_data = response.json()
            print("--- 更新失敗 (404 Not Found) ---")
            print(f"  エラーメッセージ: {error_data.get('error', 'サーバーからのエラー詳細なし')}")
            print(f"  リクエストしたID: {error_data.get('requested_id', device_id)}")
        elif response.status_code == 400: 
            error_data = response.json()
            print("--- 更新失敗 (400 Bad Request) ---")
            print(f"  エラーメッセージ: {error_data.get('error', 'サーバーからのエラー詳細なし')}")
        else:
            print(f"--- エラー (ステータスコード: {response.status_code}) ---")
            try:
                error_details = response.json()
                print("  サーバーからの詳細:")
                print(json.dumps(error_details, indent=2, ensure_ascii=False))
            except JSONDecodeError:
                print(f"  サーバーからの応答はJSON形式ではありませんでした。内容: {response.text[:200]}...")
    # ... (ConnectionError, Timeout, RequestException のキャッチはVol.6 と同じ)
    except requests.exceptions.ConnectionError:
        print(f"エラー: サーバー ({target_url}) への接続に失敗しました。")
    except requests.exceptions.Timeout:
        print(f"エラー: サーバー ({target_url}) への接続がタイムアウトしました。")
    except requests.exceptions.RequestException as e:
        print(f"エラー: リクエスト中に問題が発生しました ({target_url}): {e}")


if __name__ == '__main__':
    print("MCPクライアント (Vol.8 APIキー認証対応) を実行します。")
    print(f"使用するAPIキー: '{API_KEY}'")

    # --- 1. 全デバイス情報を取得 (認証付き) ---
    get_all_device_infos()

    # --- 2. 特定デバイス情報を取得 (認証付き) ---
    get_specific_device_info("THERMO-001-A")
    get_specific_device_info("NON-EXISTENT-DEVICE") # 存在しないID (404になるはず)

    # --- 3. 新しいデバイスを登録 (認証付き) ---
    new_smart_plug = {
        "modelName": "Smart Plug P100",
        "deviceId": "PLUG-005-E", 
        "location": "Office",
        "status": {"isOn": False, "powerConsumption": 0.5, "unit": "watts", "isActive": True},
        "supportedModes": ["on", "off"]
    }
    add_new_device(new_smart_plug)
    
    # --- 4. 既存デバイスを更新 (認証付き) ---
    updated_thermostat_info = {
        "modelName": "Smart Thermostat X1000",
        "deviceId": "THERMO-001-A", # IDは同じ
        "location": "Living Room (Updated)", # 場所を変更
        "status": {"currentTemperature": 22.0, "targetTemperature": 22.5, "unit": "Celsius", "isActive": True, "mode": "heat"},
        "supportedModes": ["auto", "cool", "heat", "off", "eco"]
    }
    update_device_info("THERMO-001-A", updated_thermostat_info)

    print("\n--- 認証テスト ---")
    print("正しいAPIキーで再度アクセスしてみます（THERMO-001-A）。")
    get_specific_device_info("THERMO-001-A") # 正しいキーで成功するはず

    # APIキーを一時的に無効なものに変えてテスト
    print("\n無効なAPIキーでアクセスを試みます...")
    original_api_key = API_KEY
    API_KEY = "WRONG_API_KEY_Haha" # わざと間違ったキーに
    get_specific_device_info("THERMO-001-A") # 401エラーになるはず
    API_KEY = original_api_key # 元に戻す

    print("\nAPIキーなしでアクセスを試みます (X-API-Keyヘッダー自体を送らない場合を模倣)...")
    # 実際にはヘッダーを送らないテストは、requestsの呼び出しで headers={} を明示的に空にするか、
    # もしくはこのテスト用にAPI_KEYをNoneにするなどして、ヘッダーが作られないようにする工夫が必要。
    # ここでは、API_KEYをNoneにしてみます。
    # (ただし、上のコードではAPI_KEYがNoneの場合、headers辞書の値がNoneになるだけで、
    # requestsライブラリがそれをどう扱うかによります。通常、値がNoneのヘッダーは送信されないか、
    # 空として送信される可能性があります。サーバー側はキーが存在し、かつ一致するかを見ているため、
    # いずれにせよ認証は失敗するはずです。)
    API_KEY = None
    print("（注意: API_KEY = None とした場合、requestsライブラリの挙動によりヘッダーが送信されないか、空で送信される可能性があります）")
    get_specific_device_info("CAM-003-C") # これも401エラーになるはず
    API_KEY = original_api_key # 忘れずに元に戻す