手を動かして学ぶ！MCPステップバイステップ実践ガイド for Beginners – Vol.9 集大成！MCP連携コマンドラインツール（一覧表示＆検索）を作る #Python

# mcp_tool.py
import argparse
import requests
import json
from requests.exceptions import JSONDecodeError # client.py から流用

# --- 設定値 (client.py Vol.8 や app.py Vol.8 と合わせる) ---
BASE_SERVER_URL = "http://127.0.0.1:5000/devices" #
API_KEY = "MySuperSecretMCPApiKey_Vol8" #
# --- ここまで設定値 ---

# --- ヘルパー関数 (client.py Vol.8 より流用・一部調整) ---
def display_device_info(device_info, indent=""):
    """デバイス情報を整形して表示するヘルパー関数"""
    print(f"{indent}モデル名: {device_info.get('modelName', '情報なし')}")
    print(f"{indent}デバイスID: {device_info.get('deviceId', '情報なし')}")
    print(f"{indent}設置場所: {device_info.get('location', '情報なし')}")

    status_info = device_info.get('status', {})
    if status_info:
        print(f"{indent}ステータス:")
        for key, value in status_info.items():
            unit = device_info.get('status', {}).get('unit', '') if key in ['currentTemperature', 'targetTemperature', 'brightness', 'powerConsumption'] else ''
            if isinstance(value, bool):
                # 'is' で始まるキーはisを除いて大文字化、それ以外はそのまま大文字化
                display_key = key.replace('is', '').capitalize() if key.startswith('is') else key.capitalize()
                print(f"{indent}  {display_key}: {'オン' if value else 'オフ'}")
            else:
                print(f"{indent}  {key.capitalize()}: {value} {unit}".strip())
    else:
        print(f"{indent}ステータス: 情報なし")

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

# --- ここまでヘルパー関数 ---

# --- コマンドごとの処理関数 ---
def handle_list_command():
    """全てのデバイス情報を取得して表示する"""
    print(f"\n--- 全てのMCPデバイス情報を取得します ---")
    target_url = BASE_SERVER_URL
    headers = {"X-API-Key": API_KEY} #

    try:
        response = requests.get(target_url, headers=headers, timeout=10) #
        response.raise_for_status()  # 4xx, 5xx エラーなら例外を発生させる

        data = response.json() #
        devices = data.get("devices", []) #

        if devices:
            print(f"--- 取得成功 ({len(devices)} 件のデバイス) ---")
            for i, device in enumerate(devices):
                print(f"\nデバイス #{i+1}")
                display_device_info(device, indent="  ")
        else:
            print("デバイス情報は見つかりませんでした。")

    except requests.exceptions.HTTPError as e: #
        print(f"エラー: HTTPエラーが発生しました。ステータスコード: {e.response.status_code}")
        try:
            error_details = e.response.json()
            print(f"  サーバーからのメッセージ: {error_details.get('message', json.dumps(error_details))}")
        except JSONDecodeError:
            print(f"  サーバーからの応答はJSON形式ではありませんでした。内容(一部): {e.response.text[:200]}...")
    except requests.exceptions.ConnectionError: #
        print(f"エラー: サーバー ({target_url}) への接続に失敗しました。サーバーが起動しているか、URLが正しいか確認してください。")
    except requests.exceptions.Timeout: #
        print(f"エラー: サーバー ({target_url}) からの応答がタイムアウトしました。")
    except JSONDecodeError: #
        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 handle_get_command(device_id):
    """指定されたIDのデバイス情報を取得して表示する"""
    print(f"\n--- MCPデバイスID '{device_id}' の情報を取得します ---")
    target_url = f"{BASE_SERVER_URL}/{device_id}" #
    headers = {"X-API-Key": API_KEY} #

    try:
        response = requests.get(target_url, headers=headers, timeout=10) #
        
        if response.status_code == 200: #
            device_data = response.json() #
            print("--- 取得成功 ---")
            display_device_info(device_data, indent="  ")
        elif response.status_code == 401: #
            error_data = response.json()
            print("--- 認証失敗 (401 Unauthorized) ---")
            print(f"  サーバーからのメッセージ: {error_data.get('message', 'APIキーが無効か、提供されていません。')}")
        elif response.status_code == 404: #
            error_data = response.json()
            print("--- 取得失敗 (404 Not Found) ---")
            print(f"  サーバーからのメッセージ: {error_data.get('message', '指定されたデバイスは見つかりませんでした。')}")
            print(f"  リクエストしたID: {device_id}")
        else: # その他のHTTPエラー
            print(f"--- エラー (ステータスコード: {response.status_code}) ---")
            response.raise_for_status() # より詳細なHTTPErrorを発生させる

    except requests.exceptions.HTTPError as e: #
        # raise_for_status() や上記のelse節から来る場合
        print(f"エラー: HTTPエラーが発生しました (上記以外)。ステータスコード: {e.response.status_code}")
        try:
            error_details = e.response.json()
            print(f"  サーバーからのメッセージ: {error_details.get('message', json.dumps(error_details))}")
        except JSONDecodeError:
            print(f"  サーバーからの応答はJSON形式ではありませんでした。内容(一部): {e.response.text[:200]}...")
    except requests.exceptions.ConnectionError: #
        print(f"エラー: サーバー ({target_url}) への接続に失敗しました。サーバーが起動しているか、URLが正しいか確認してください。")
    except requests.exceptions.Timeout: #
        print(f"エラー: サーバー ({target_url}) からの応答がタイムアウトしました。")
    except JSONDecodeError: #
        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}")

# --- argparse を使ったコマンドライン引数の処理 ---
if __name__ == '__main__':
    # 1. メインのパーサーを作成
    parser = argparse.ArgumentParser(
        description="MCPデバイス情報を操作するためのコマンドラインツールです。",
        epilog="使用例: python mcp_tool.py list"
    )
    # バージョン情報 (任意)
    parser.add_argument('--version', action='version', version='%(prog)s 1.0')

    # 2. サブパーサー（サブコマンド）を格納するコンテナを作成
    # dest="command" で選択されたサブコマンド名が格納される
    # required=True でサブコマンドの指定を必須にする (Python 3.7+)
    subparsers = parser.add_subparsers(dest="command", title="利用可能なコマンド", help="実行するサブコマンドを選んでください。", required=True)

    # 3. 'list' コマンド用のサブパーサーを作成
    list_parser = subparsers.add_parser(
        "list", 
        help="全てのMCPデバイス情報を一覧表示します。",
        description="サーバーに登録されている全てのMCPデバイスの情報を取得し、整形して表示します。"
    )
    # 'list' コマンドに引数が必要な場合はここに追加する (今回はなし)
    # list_parser.add_argument(...) 

    # 4. 'get' コマンド用のサブパーサーを作成
    get_parser = subparsers.add_parser(
        "get",
        help="指定したIDのMCPデバイス情報を取得します。",
        description="指定されたデバイスIDに対応するMCPデバイスの情報を取得し、詳細を表示します。"
    )
    get_parser.add_argument(
        "--device-id",  # オプション名 (ハイフン2つで始まるのが慣例)
        metavar="DEVICE_ID", # ヘルプメッセージで表示される引数の名前
        required=True,    # この引数を必須にする
        help="取得したいデバイスのIDを指定してください。"
    )

    # 5. コマンドライン引数をパース
    args = parser.parse_args()

    # 6. パースされた引数に基づいて処理を分岐
    if args.command == "list":
        handle_list_command()
    elif args.command == "get":
        handle_get_command(args.device_id)
    else:
        # required=True なので通常ここには来ないが、念のため
        print("不明なコマンドです。--help を参照してください。")