8
3

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

手を動かして学ぶ!MCPステップバイステップ実践ガイド for Beginners - Vol.9 集大成!MCP連携コマンドラインツール(一覧表示&検索)を作る

Posted at

はじめに

皆さん、こんにちは!「手を動かして学ぶ!MCPステップバイステップ実践ガイド for Beginners」へようこそ。いよいよこのシリーズも佳境に入ってきました。

前回 (Vol.8 秘密の合言葉!APIキーでMCPアクセスに認証を追加する) は、APIキーを使った認証の仕組みを導入し、私たちのMCPサーバーのセキュリティを高めましたね。 これで、許可されたクライアントだけがサーバーとやり取りできるようになりました。

今回は、まさに「集大成」として、これまで学んできたこと全てを活かして、実用的な「コマンドラインツール」を作成します。このツールを使えば、ターミナル(黒い画面でおなじみの、コマンドを打ち込むアレです)から手軽にMCPデバイスの情報を一覧表示したり、特定のデバイス情報を検索(取得)したりできるようになります。

これまでは client.py というスクリプトを実行してきましたが、今回はより本格的なツールとして、コマンドの引数(オプション)を使って動作を切り替えられるようにします。まるで、私たちが普段使っている lsgit のようなコマンドラインツールを自分で作るようなイメージです。これまで一つ一つ作ってきた部品を組み合わせて、便利な道具を作り上げる楽しさを味わいましょう!

1. コマンドラインツールって何? 💻

コマンドラインツール(Command-Line Tool)とは、グラフィカルなボタンやウィンドウ(GUI: Graphical User Interface)ではなく、キーボードからコマンド(命令)を打ち込んで操作するタイプのソフトウェアです。CLI (Command-Line Interface) とも呼ばれます。

皆さんも、知らず知らずのうちに使っているかもしれません。例えば、

  • ls (Linux/macOS) や dir (Windows): ファイルやディレクトリの一覧を表示するコマンド
  • cd: ディレクトリを移動するコマンド
  • git: バージョン管理システムGitを操作するコマンド
  • pipnpm: プログラミング言語のパッケージを管理するコマンド

これらは全てコマンドラインツールです。GUIに比べて見た目はシンプルですが、以下のようなメリットがあります。

  • 自動化しやすい: スクリプトと組み合わせて定型的な作業を自動化するのに向いています。
  • 効率的な操作: 慣れるとマウス操作よりも素早く多くの作業を行えます。
  • リソース消費が少ない: GUIに比べてコンピュータのメモリやCPUへの負荷が小さいことが多いです。
  • サーバー環境での利用: GUIがないサーバー環境でも利用できます。

今回は、Pythonの標準ライブラリである argparse (アーグパース)モジュールを使って、このコマンドラインツールを作成します。argparse は、コマンドラインから渡される引数(「オプション」や「パラメータ」とも呼ばれます)を簡単に扱えるようにし、使い方を示すヘルプメッセージなども自動で生成してくれる便利なモジュールです。

2. 設計しよう!どんなツールにする? 🛠️

まずは、私たちが作るMCPコマンドラインツールの機能を設計しましょう。

機能要件

最低限、以下の2つの機能を持つツールを目指します。

1. デバイス一覧表示機能: サーバーに登録されている全てのMCPデバイス情報を取得し、一覧で表示する。

  • コマンド例: python mcp_tool.py list

2. 特定デバイス情報表示機能 (IDによる検索): 指定した deviceId を持つMCPデバイスの情報を取得し、表示する。これを「検索」機能の一つと捉えます。

  • コマンド例: python mcp_tool.py get --device-id THERMO-001-A

今回は、より複雑な条件での「検索」(例: モデル名の一部で検索など)はサーバー側の対応も必要になるため、まずはID指定での情報取得を「検索」の第一歩とします。

共通の要素

  • サーバーURL: 接続先のMCPサーバーのURL。今回はスクリプト内に固定値として持ちます。
  • APIキー: 認証のためのAPIキー。こちらもスクリプト内に固定値として持ちます。(実際のツールでは設定ファイルや環境変数から読み込むのがより安全で柔軟です。)
  • エラーハンドリング: 通信エラーや認証エラーなど、前回までに学んだエラー処理をしっかり組み込みます。
  • 出力形式: デバイス情報を見やすく表示します。Vol.8までの display_device_info 関数を参考にします。

3. 実装しよう!MCPコマンドラインツール (mcp_tool.py) を作る 👨‍💻👩‍💻

それでは、実際にPythonコードを書いていきましょう。新しいファイル mcp_tool.py を作成してください。

必要なもの・環境設定

  • Python 3.x
  • requests ライブラリ (pip install requests または pip3 install requests)

argparse はPythonの標準ライブラリなので、追加のインストールは不要です。

mcp_tool.py の基本構造

# 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 を参照してください。")

コードのポイント:

  • 設定値: BASE_SERVER_URLAPI_KEY は、Vol.8で作成したサーバー (app.py) とクライアント (client.py) の設定と合わせてください。
  • ヘルパー関数 display_device_info: Vol.8client.py から流用し、インデントを調整できるように引数 indent を追加しました。
  • コマンド処理関数:
    • handle_list_command(): 全デバイス情報を取得・表示します。内部のロジックは client.pyget_all_device_infos() とほぼ同じです。
    • handle_get_command(device_id): 指定されたIDのデバイス情報を取得・表示します。内部のロジックは client.pyget_specific_device_info() とほぼ同じです。
    • どちらの関数も、APIキーをヘッダーに含めてリクエストを送信し、適切なエラーハンドリングを行っています。
  • argparse の設定:
    • parser = argparse.ArgumentParser(...): メインとなるパーサーオブジェクトを作成します。descriptionepilog でツールの説明を追加できます。
    • subparsers = parser.add_subparsers(...): listget といったサブコマンドを定義するためのものです。dest="command" で、実行されたサブコマンドの名前が args.command に格納されるようになります。required=True を指定することで、サブコマンドのいずれかが必ず指定されるようにします (Python 3.7以降)。
    • list_parser = subparsers.add_parser("list", ...): list サブコマンド用のパーサーを定義します。
    • get_parser = subparsers.add_parser("get", ...): get サブコマンド用のパーサーを定義します。
    • get_parser.add_argument("--device-id", ...): get コマンドが受け付ける引数 --device-id を定義します。required=True でこの引数を必須にしています。
  • メイン処理 (if __name__ == '__main__':):
    • args = parser.parse_args(): コマンドラインから渡された引数を解析し、args オブジェクトに格納します。
    • if args.command == "list": ...: args.command の値に応じて、対応する処理関数を呼び出します。

これで、mcp_tool.py という一つのファイルで、複数の操作を行えるコマンドラインツールが完成しました!

4. 使ってみよう!コマンドラインツールの実行 🚀

サーバー (app.py Vol.8版) が起動していることを確認してから、作成した mcp_tool.py を実行してみましょう。

準備:

Vol.8で作成・使用した app.py を起動しておきます(APIキーが MySuperSecretMCPApiKey_Vol8 になっているもの)。

python app.py

(macOSやLinuxで python3 を使っている方は python3 app.py)

コマンド実行例:

1. ヘルプメッセージの表示:
まず、ツール全体のヘルプを見てみましょう。

python mcp_tool.py --help

(macOSやLinuxで python3 を使っている方は python3 mcp_tool.py --help

すると、argparse が自動生成したツールの説明や利用可能なコマンド一覧が表示されます。

各サブコマンドのヘルプも確認できます。

python mcp_tool.py list --help
python mcp_tool.py get --help

2. 全デバイス一覧表示:

python mcp_tool.py list

サーバーに登録されているデバイス情報が一覧で表示されれば成功です。

3. 特定デバイスの情報表示 (ID指定):
Vol.8app.py にデフォルトで登録されているデバイスID THERMO-001-A を指定してみましょう。

python mcp_tool.py get --device-id THERMO-001-A

THERMO-001-A の詳細情報が表示されれば成功です。

4. 存在しないデバイスIDを指定:

python mcp_tool.py get --device-id NON_EXISTENT_ID_XYZ

「取得失敗 (404 Not Found)」や「指定されたデバイスは見つかりませんでした。」といったエラーメッセージが表示されれば、エラー処理も正しく機能しています。

5. (おまけ) APIキーが違う場合:
もし mcp_tool.py 内の API_KEY を一時的に間違ったものに変更して listget を実行すると、「認証失敗 (401 Unauthorized)」のエラーメッセージが表示されるはずです。これもAPIキー認証が機能している証拠ですね。(試した後は正しいキーに戻しておきましょう。)

5. まとめと次回予告

今回は、これまでに学んだMCP通信の知識、エラー処理、APIキー認証などを組み合わせて、実用的なコマンドラインツール mcp_tool.py を作成しました。argparse モジュールを使うことで、本格的なコマンドライン引数の処理も実現できましたね。

今回の学びのポイント:

  • コマンドラインツールの基本: コマンドと引数で操作するツールの概念。
  • argparse モジュール: Pythonでコマンドラインツールを作る際の強力な味方。サブコマンドや引数の定義、ヘルプメッセージの自動生成など。
  • 機能の組み合わせ: 既存の関数やロジック(HTTPリクエスト、JSON処理、エラーハンドリング、認証ヘッダーの付与など)を再利用・統合して新しいツールを作ることの体験。
  • ツールの設計: どのような機能を持ち、ユーザーがどのように操作するかを考えることの重要性。

作成した mcp_tool.py は、さらに機能を拡張していくことも可能です。例えば、

  • 新しいデバイスを登録する add コマンド
  • 既存デバイス情報を更新する update コマンド
  • デバイスを削除する delete コマンド (サーバー側の実装も必要)
  • より柔軟な検索機能(例: モデル名や設置場所での部分一致検索など。これもサーバー側の対応が必要)
  • APIキーやサーバーURLを設定ファイルから読み込む機能

などを追加していくと、さらに便利なツールになるでしょう。

さて、長かったこの「手を動かして学ぶ!MCPステップバイステップ実践ガイド for Beginners」シリーズも、いよいよ次回で最終回です。

Vol.10 のテーマは「完走!MCP実践の振り返りと、ここから始まる学びの道」です。シリーズ全体を振り返り、私たちが何を学び、何ができるようになったかを確認します。そして、この経験を活かして次にどんなことを学んでいけるか、さらなるステップアップのためのヒントや学習リソースについても触れたいと思います。

最後までお付き合いいただき、ありがとうございます。最終回もどうぞお楽しみに!

8
3
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
8
3

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?