はじめに
皆さん、こんにちは！「手を動かして学ぶ！MCPステップバイステップ実践ガイド for Beginners」へようこそ。このシリーズでは、Model Context Protocol (MCP) という仕組みを、Pythonを使って実際に手を動かしながら学んでいます。
前回 (Vol.7 転ばぬ先の杖！MCP通信のエラー処理 (try...except) を学ぼう) は、プログラムの安定性を高めるために、エラーが発生したときに適切に対処する「エラー処理」について学びましたね。try...exceptブロックやサーバーからの適切なエラー応答の重要性を理解いただけたかと思います。
さて、これまでの私たちのMCPサーバーは、誰でも自由にアクセスできてしまう状態でした。個人の学習用であればそれでも問題ありませんが、実際のWebサービスやAPIでは、「誰でも彼でも使っていいよ」というわけにはいきません。特定の許可された人やプログラムだけがサービスを利用できるように制限する必要があります。これが「認証 (Authentication)」という考え方です。
今回は、この認証の仕組みの一つとして比較的シンプルで広く使われている「APIキー認証」を導入します。まるで秘密のクラブに入るための「合言葉」のように、クライアントとサーバーの間でだけ通用する「APIキー」を使って、MCPサーバーへのアクセスを制御する方法を一緒に見ていきましょう！
1. APIキー認証ってなに？ 🔑
APIキー認証とは、文字通り「APIキー」と呼ばれる特別な文字列（合言葉）を使って、API（私たちの場合はMCPサーバー）の利用者を認証する仕組みです。
APIキー認証の仕組み
-
サーバーがAPIキーを発行・管理:
サーバーの管理者があらかじめ、正当なクライアントに対してユニークなAPIキー（例:abc123xyz789のようなランダムな文字列）を発行します。サーバーはこの「有効なAPIキーのリスト」を保持しておきます。 -
クライアントがリクエスト時にAPIキーを提示:
クライアントは、サーバーにリクエストを送る際に、発行された自分のAPIキーを一緒に送信します。 -
サーバーがAPIキーを検証:
サーバーは、リクエストを受け取ると、まず送られてきたAPIキーが有効なものかどうかを確認します。
- 有効な場合: 「この人は許可された利用者だな」と判断し、リクエストされた処理を実行します。
- 無効または未提示の場合: 「あなたは誰ですか？許可されていませんよ」と判断し、リクエストを拒否します（通常、エラーメッセージを返します）。
まるで、会員制クラブの入り口で、会員証（APIキー）を見せて中に入るようなイメージですね。会員証を持っていなければ、門前払いされてしまいます。
APIキーはどこで送るの？
クライアントがサーバーにAPIキーを送る方法はいくつかありますが、一般的に使われるのは以下の方法です。
-
HTTPヘッダー (Custom HTTP Header): HTTPリクエストの「ヘッダー」と呼ばれるメタ情報部分に、
X-API-Key: YOUR_API_KEYのように独自のヘッダー名でAPIキーを含める方法です。これは現在最も一般的で推奨される方法の一つです。リクエストの本体（ボディ）とは別に、付加情報として送信できます。 （例:X-API-Key、Authorization: Bearer YOUR_API_KEY※後者はより汎用的な認証ヘッダー） -
クエリパラメータ (Query Parameter): URLの末尾に
?api_key=YOUR_API_KEYのように付けて送信する方法です。手軽ですが、URLにAPIキーが含まれるため、ブラウザの履歴やサーバーのアクセスログに残りやすく、セキュリティ面で若干不利になることがあります。 - リクエストボディ (Request Body): POSTリクエストなどのボディ部分にAPIキーを含める方法もありますが、認証情報としてはヘッダーで送る方が一般的です。
今回は、最も標準的で安全性の高い方法の一つである「カスタムHTTPヘッダー」を使ってAPIキーを送受信する方法を実装します。
2. サーバー側の門番：APIキーをチェックする (app.pyの改造) 🛡️
まずは、MCPサーバー(app.py)に「門番」の役割を持たせましょう。クライアントから送られてくるAPIキーをチェックし、正しいキーを持つリクエストだけを通すようにします。
必要なもの・環境設定
- Python 3.x
- Flask ライブラリ (
pip install Flaskまたはpip3 install Flask) （これらはVol.7までと同様です。）
app.py の変更点 (Vol.8版)
Vol.7の app.py をベースに、以下の変更を加えます。
-
有効なAPIキーの定義:
サーバー側で「これが正解のAPIキーだよ」という値を定義します。今回はわかりやすく説明するため、Pythonのコード内に直接書き込みます。実際の運用では、設定ファイルや環境変数から読み込むことが多いです。 -
APIキーをチェックするデコレータの作成:
Pythonの「デコレータ (decorator)」という機能を使って、APIキーのチェック処理を部品化します。デコレータは、既存の関数に前処理や後処理を簡単に追加できる便利な仕組みです。「関数のラッパー」とも呼ばれます。 このデコレータは、リクエストヘッダーからAPIキーを取得し、それが有効かどうかを判定します。無効な場合は「401 Unauthorized」というエラー（認証されていませんよ、という意味）を返します。 -
ルートへのデコレータの適用:
作成したAPIキーチェックデコレータを、保護したいルート（エンドポイント）に適用します。これにより、それらのルートはAPIキーがないとアクセスできなくなります。
app.py (Vol.8 版)
# app.py (Vol.8 版)
from flask import Flask, jsonify, request
import logging
from functools import wraps # デコレータ作成のためにインポート
app = Flask(__name__)
# --- ロガー設定 (Vol.7 と同じ) ---
if not app.debug:
stream_handler = logging.StreamHandler()
stream_handler.setLevel(logging.INFO)
app.logger.addHandler(stream_handler)
app.logger.setLevel(logging.INFO)
# --- ここまでロガー設定 ---
# ★★★ 新しい変更点：有効なAPIキーを定義 ★★★
VALID_API_KEY = "MySuperSecretMCPApiKey_Vol8" # 例：これがサーバーの知っている「秘密の合言葉」
# ★★★ 新しい変更点：APIキーをチェックするデコレータ ★★★
def require_api_key(f):
@wraps(f) # これにより、デコレートされた関数の名前などが元の関数から引き継がれる
def decorated_function(*args, **kwargs):
api_key_received = request.headers.get("X-API-Key") # クライアントは 'X-API-Key' ヘッダーで送ってくる想定
if api_key_received and api_key_received == VALID_API_KEY:
# APIキーが提供され、かつ有効なキーと一致する場合
return f(*args, **kwargs) # 元の関数（ルート処理）を実行
else:
# APIキーがない、または無効な場合
app.logger.warning(
f"Unauthorized access attempt to {request.path}. "
f"Provided API Key: '{api_key_received if api_key_received else 'Not provided'}'"
)
return jsonify(error="Unauthorized", message="有効なAPIキーが提供されていません。アクセスが拒否されました。"), 401 # 401 Unauthorized エラーを返す
return decorated_function
# --- サンプルのMCPデバイスデータ (Vol.7 と同じ) ---
all_devices_data = [
# ... (内容はVol.7 と同じなので省略) ...
{
"modelName": "Smart Thermostat X1000",
"deviceId": "THERMO-001-A",
"location": "Living Room",
"status": {"currentTemperature": 23.5, "targetTemperature": 24.0, "unit": "Celsius", "isActive": True, "mode": "auto"},
"supportedModes": ["auto", "cool", "heat", "off"]
},
{
"modelName": "Smart Light L200",
"deviceId": "LIGHT-002-B",
"location": "Bedroom",
"status": {"brightness": 80, "color": "warm_white", "unit": "percent", "isActive": True},
"supportedModes": ["on", "off", "dim"]
},
{
"modelName": "Security Camera C300",
"deviceId": "CAM-003-C",
"location": "Entrance",
"status": {"isRecording": False, "detectionMode": "motion", "isActive": True},
"supportedEvents": ["motion_detected", "sound_detected"]
}
]
# --- ここまでサンプルデータ ---
# 全デバイス情報を返すエンドポイント (GET) と新しいデバイスを登録するエンドポイント (POST)
@app.route('/devices', methods=['GET', 'POST'])
@require_api_key # ★★★ 変更点：作成したデコレータを適用 ★★★
def handle_devices():
# この関数内の処理はVol.7 と基本的に同じ (APIキーチェックはデコレータが担当)
if request.method == 'POST':
new_device_data = request.json
if not new_device_data:
app.logger.warning("POST /devices - No data provided by client.")
return jsonify({"error": "No data provided", "message": "リクエストボディにデータが含まれていません。"}), 400
new_device_id = new_device_data.get("deviceId")
if not new_device_id:
app.logger.warning(f"POST /devices - deviceId is missing. Data: {new_device_data}")
return jsonify({"error": "deviceId is required", "message": "必須フィールド 'deviceId' がありません。"}), 400
for device in all_devices_data:
if device.get("deviceId") == new_device_id:
app.logger.info(f"POST /devices - Device with ID {new_device_id} already exists.")
return jsonify({"error": "Device with this ID already exists", "message": f"デバイスID '{new_device_id}' は既に存在します。", "deviceId": new_device_id}), 409
all_devices_data.append(new_device_data)
app.logger.info(f"POST /devices - New device created: {new_device_id}")
return jsonify(new_device_data), 201
# GET リクエスト
app.logger.info("GET /devices - Returning all devices.")
return jsonify({"devices": all_devices_data})
# 特定のデバイスIDに基づいて情報を返す (GET) / 更新する (PUT) エンドポイント
@app.route('/devices/<device_id>', methods=['GET', 'PUT'])
@require_api_key # ★★★ 変更点：作成したデコレータを適用 ★★★
def handle_specific_device(device_id):
# この関数内の処理はVol.7 と基本的に同じ (APIキーチェックはデコレータが担当)
found_device_index = -1
for i, device in enumerate(all_devices_data):
if device.get("deviceId") == device_id:
found_device_index = i
break
if request.method == 'PUT':
if found_device_index != -1:
updated_data = request.json
if not updated_data:
app.logger.warning(f"PUT /devices/{device_id} - No data provided for update.")
return jsonify({"error": "No data provided for update", "message": "更新データがリクエストボディに含まれていません。"}), 400
original_device_id_in_payload = updated_data.get("deviceId")
# ... (deviceIdミスマッチ処理などはVol.7 と同様) ...
if original_device_id_in_payload and original_device_id_in_payload != device_id:
all_devices_data[found_device_index] = updated_data
app.logger.info(f"PUT /devices/{device_id} - Device updated. Payload deviceId '{original_device_id_in_payload}' differed from URL.")
return jsonify({"warning": "Device ID in URL and payload mismatch. Resource updated based on URL.", "message": f"デバイスID '{device_id}' の情報が更新されましたが、リクエストデータ内のdeviceId ('{original_device_id_in_payload}') はURLと一致しませんでした。URLのIDに基づいて処理されました。", "updated_device": updated_data}), 200
all_devices_data[found_device_index] = updated_data
app.logger.info(f"PUT /devices/{device_id} - Device updated successfully.")
return jsonify(updated_data), 200
else:
app.logger.warning(f"PUT /devices/{device_id} - Device not found for update.")
return jsonify({"error": "Device not found, cannot update", "message": f"デバイスID '{device_id}' は見つからなかったため、更新できませんでした。", "requested_id": device_id}), 404
# GETリクエストの場合
if found_device_index != -1:
app.logger.info(f"GET /devices/{device_id} - Returning device information.")
return jsonify(all_devices_data[found_device_index])
else:
app.logger.warning(f"GET /devices/{device_id} - Device not found.")
return jsonify({"error": "Device not found", "message": f"デバイスID '{device_id}' は見つかりませんでした。", "requested_id": device_id}), 404
# --- エラーハンドラ (Vol.7 と同じ) ---
@app.errorhandler(404)
def resource_not_found(e):
app.logger.warning(f"Unhandled route or resource not found: {request.path} - {e}")
return jsonify(error="Not Found", message="お探しのリソースは見つかりませんでした。URLを確認してください。"), 404
@app.errorhandler(500)
def internal_server_error(e):
app.logger.error(f"Server Error: {e}", exc_info=True)
return jsonify(error="Internal Server Error", message="申し訳ありません、サーバー内部で予期せぬエラーが発生しました。管理者にお問い合わせください。"), 500
@app.errorhandler(405)
def method_not_allowed(e):
app.logger.warning(f"Method Not Allowed: {request.method} for {request.path} - {e}")
return jsonify(error="Method Not Allowed", message=f"メソッド '{request.method}' はこのURLでは許可されていません。"), 405
@app.errorhandler(400)
def bad_request_error(e):
app.logger.warning(f"Bad Request: {request.path} - {e}")
detail_message = e.description if hasattr(e, 'description') else "リクエストの形式が正しくありません。"
return jsonify(error="Bad Request", message=detail_message), 400
# --- ここまでエラーハンドラ ---
if __name__ == '__main__':
print("MCP Server (Vol.8 with API Key Auth) is running on http://127.0.0.1:5000")
print(f"Make sure to send the API Key '{VALID_API_KEY}' in the 'X-API-Key' header.")
print("To stop the server, press CTRL+C")
# macOSやLinuxで python3 を使っている場合は、python3 app.py で実行してください。
app.run(debug=True, port=5000)
コードのポイント:
-
from functools import wraps: デコレータを正しく作るためのおまじないのようなものです。これを使うと、デコレータでラップされた関数が、元の関数の情報（名前など）を保持しやすくなります。 -
VALID_API_KEY = "MySuperSecretMCPApiKey_Vol8": ここで「正解のAPIキー」を定義しています。この文字列と一致するキーがクライアントから送られてくればOKです。 -
def require_api_key(f): ...: これがAPIキーチェックデコレータの定義です。-
@wraps(f): 上述のおまじない。 -
def decorated_function(*args, **kwargs): ...: 実際にリクエストを処理する前に行われるチェック処理を書く関数です。 -
request.headers.get("X-API-Key"): リクエストのヘッダーからX-API-Keyという名前のヘッダーの値を取得します。クライアントがこのヘッダー名でAPIキーを送ってくることを期待しています。 - キーが正しい場合は
return f(*args, **kwargs)で元のルート関数（handle_devicesやhandle_specific_device）を呼び出し、その結果を返します。 - キーが不正またはない場合は、
jsonify(...)でエラーメッセージと401ステータスコードを返します。この時点で処理は中断され、元のルート関数は呼び出されません。
-
-
@require_api_key: これを保護したいルート関数の直前に置くことで、そのルート関数が呼び出される前に必ずrequire_api_keyデコレータ内のチェック処理が実行されるようになります。 - 既存のルート関数
handle_devicesとhandle_specific_deviceの中身は、APIキー認証に関するロジックを直接書く必要はありません。デコレータがその役割を担ってくれるため、ルート関数自体は本来の処理に集中できます。これがデコレータの強力な点です。 -
401 Unauthorizedエラー: APIキーが無効な場合にこのステータスコードを返すのは、HTTPの標準的な慣習です。「認証が必要ですよ、または認証に失敗しましたよ」という意味合いです。
これでサーバー側の準備は完了です。門番が立って、合言葉を知らない人は通れなくなりました。
3. クライアントからの合言葉：APIキーを送る (client.pyの改造) 🗣️
次に、クライアント側(client.py)が、この「合言葉」であるAPIキーをサーバーに正しく伝えられるように改造しましょう。
client.py の変更点 (Vol.8版)
Vol.6の client.py (エラー処理などが含まれたバージョン) をベースに、以下の変更を加えます。
-
APIキーの定義:
クライアント側でも、使用するAPIキーを定義します。この値は、サーバー側のVALID_API_KEYと完全に一致している必要があります。 -
リクエストヘッダーへのAPIキーの追加:
requestsライブラリを使ってHTTPリクエストを送信する際に、headers引数を使ってカスタムヘッダー (X-API-Key) にAPIキーを設定します。これを、情報を取得・登録・更新する全ての関数で行います。
client.py (Vol.8 版)
# 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 # 忘れずに元に戻す
コードのポイント:
-
API_KEY = "MySuperSecretMCPApiKey_Vol8": クライアントが使用するAPIキーを定義します。 -
headers = {"X-API-Key": API_KEY}:requestsライブラリでリクエストを送信する際に、このheaders辞書を渡します。これにより、HTTPリクエストにX-API-Key: MySuperSecretMCPApiKey_Vol8というヘッダーが付加されて送信されます。 - 各関数 (
get_all_device_infos,get_specific_device_info,add_new_device,update_device_info) のrequests.get()、requests.post()、requests.put()の呼び出し部分にheaders=headersが追加されています。 - クライアント側のエラー処理 (
try...exceptブロック) は基本的にVol.6 のものを流用していますが、サーバーから401 Unauthorizedが返ってきた場合に、そのメッセージを適切に表示するように少し調整しています（特にget_specific_device_infoやadd_new_deviceなどで、elif response.status_code == 401:のような分岐を追加または明確化）。 -
if __name__ == '__main__':ブロックでは、正しいAPIキーを使った場合の動作確認と、意図的に間違ったAPIキーやキーなし（API_KEY = Noneとして模倣）の場合のテストを追加しています。
4. 動かしてみよう！認証の動作確認 🎬
サーバーとクライアントの準備が整いました。実際に動かして、APIキー認証が正しく機能するかを確認しましょう。
手順:
1. サーバーの起動:
app.py (Vol.8 版) があるディレクトリで、ターミナル（またはコマンドプロンプト）を開き、以下のコマンドを実行します。
python app.py
（macOSやLinuxの方は python3 app.py）
サーバーが起動し、Make sure to send the API Key 'MySuperSecretMCPApiKey_Vol8' in the 'X-API-Key' header. のようなメッセージが表示されるはずです。
2. クライアントの実行:
別の ターミナル（またはコマンドプロンプト）を開き、client.py (Vol.8 版) があるディレクトリで、以下のコマンドを実行します。
python client.py
（macOSやLinuxの方は python3 client.py）
期待される結果:
-
正しいAPIキーの場合:
client.pyのAPI_KEYがサーバーのVALID_API_KEYと一致していれば、これまでのVol.6やVol.7と同様に、デバイス情報の取得、登録、更新が成功するはずです。 -
client.pyのif __name__ == '__main__':内のテスト部分:- 「無効なAPIキーでアクセスを試みます...」の箇所では、クライアントの出力に「認証失敗 (401 Unauthorized)」「サーバーからのメッセージ: 有効なAPIキーが提供されていません。アクセスが拒否されました。」といったエラーが表示されるはずです。
- サーバー側のターミナルには、
Unauthorized access attempt ... Provided API Key: 'WRONG_API_KEY_Haha'のような警告ログが出力されます。 - 「APIキーなしでアクセスを試みます...」の箇所でも同様に、認証失敗のメッセージと、サーバーログにキーが提供されなかった旨の記録が残るはずです。
このように、APIキーが正しくなければサーバーがリクエストを拒否し、クライアント側でもそのエラーを適切にハンドリングできていることが確認できれば成功です。
5. APIキー認証のポイントとこれから
今回実装したAPIキー認証は、比較的簡単に導入できる認証方法ですが、いくつかの注意点と限界も理解しておきましょう。
セキュリティに関する注意点:
- APIキーの機密性: APIキーはパスワードと同じように秘密に保つ必要があります。絶対に公開リポジトリ（GitHubのパブリックな場所など）に直接書き込んだり、安全でない方法で共有したりしないでください。
- HTTPS通信の利用: 現在の私たちの通信はHTTPで行われており、通信経路上でAPIキーが盗聴される可能性があります。実際の運用では、必ずHTTPS（暗号化されたHTTP）を使って通信し、APIキーを保護する必要があります。（このシリーズではHTTPSのセットアップは範囲外としますが、非常に重要です。）
- キーの管理: APIキーが漏洩した場合に備えて、キーを無効化したり再発行したりする仕組みを検討することも重要です。複数のクライアントに異なるキーを発行することで、問題が発生したクライアントのキーだけを無効にするといった対応も可能になります。
APIキー認証の限界:
APIキー認証は、クライアント（アプリケーション）を認証するのには役立ちますが、個々の「ユーザー」を認証するのには向いていません。より複雑なユーザー認証や認可（特定の操作に対する権限管理）が必要な場合は、OAuth 2.0のような、より高度な認証・認可の仕組みを検討する必要があります。
とはいえ、サーバー間の通信や、信頼できるクローズドな環境でのクライアント認証としては、APIキー認証は手軽で有効な手段です。
さて、これで私たちのMCPサーバーには「認証」というセキュリティ機能が備わりました。次回、Vol.9 のテーマは 「集大成！MCP連携コマンドラインツール（一覧表示＆検索）を作る」 です。
これまで学んできた知識（GET, POST, PUT, エラー処理, APIキー認証）を総動員して、実際にコマンドラインから手軽にMCPデバイス情報を操作できるツールを作成していきます。これまでの学習の成果を形にする、楽しみな回です！
どうぞお楽しみに！