セキュリティ、信頼性、オブザーバビリティ

imgimgimgimgimg
このレッスンのロックを解除するには、まず「MCPとAIエージェントの接続」を完了してください。前のレッスンに戻る

MCP 統合におけるセキュリティ境界

MCP 統合は、AI が制御する意思決定プロセスを、サーバーが公開するツール、リソース、prompt に接続します。この接続は、単なる API の利便性ではなく、セキュリティ境界です。モデルが生成したツール引数は、信頼できない入力として扱う必要があります。モデルがリクエストを誤解したり、悪意のあるコンテンツに従ったり、意図しない結果を招くツールを選択したりする可能性があるためです。MCP サーバーから返されるデータも信頼できません。機密情報、不正な構造、または後続のモデルの動作に影響を与えるよう設計された指示が含まれている可能性があります。

有用な脅威モデルでは、ユーザー、ホストアプリケーション、モデル、MCP クライアント、MCP サーバー、下流システムを分離して考えます。認証はアイデンティティを確立し、認可はそのアイデンティティに許可される操作を決定します。モデルにアプリケーションプロセスが持つすべての権限を継承させてはいけません。たとえば、サポートアシスタントは顧客レコードを読み取れても、無制限のデータベースアクセス権や、追加のポリシーチェックなしに返金を実行する権限を与えられてはなりません。

信頼はサーバー全体ではなく、個々の操作に対して付与すべきです。1 つのサーバーが、安全な商品検索ツールと、破壊的なアカウント削除ツールを併せて公開している場合があります。アプリケーションに必要な機能だけを登録し、影響度によって分類するとともに、現在のワークフローで必要になるまで影響の大きいツールを利用できないようにします。金銭、アクセス権、法的ステータス、本番データ、または外部とのコミュニケーションを変更する操作では、人による承認が適切です。

ユーザー、AI ホスト、MCP クライアント、MCP サーバー、下流サービスを示し、ツール実行前の検証および認可ゲートと、戻り経路上の出力フィルタリングを備えた信頼境界図

検証と最小権限のアクセス許可

MCP ツール定義ですでに入力スキーマが記述されている場合でも、実行直前にツール引数を検証します。説明はモデルがリクエストを組み立てる際に役立ちますが、アプリケーションポリシーを強制するものではありません。検証では、必須フィールドの欠落、想定外のフィールド、誤った型、過剰な長さ、不正な識別子、許可範囲外の値を拒否する必要があります。構造上の制約と同じく、ビジネス上の制約も重要です。構文的に正しい返金額であっても、現在のユーザーの承認上限を超える可能性があります。

次の実行可能な Python の例では、仮想的な MCP ツールアダプターの周囲に小規模なセキュリティ境界を実装します。標準ライブラリのみを使用します。アダプター関数は、実際の MCP クライアントが SDK またはトランスポート経由で実行する呼び出しの代わりとなるものです。ゲートウェイは引数を検証し、呼び出し元の権限を確認し、状態を変更する操作には承認を要求します。

python
from dataclasses import dataclass
from typing import Any, Callable


@dataclass(frozen=True)
class Principal:
    user_id: str
    permissions: frozenset[str]


TOOL_POLICY = {
    "customer.read": {"permission": "customer:read", "mutating": False},
    "refund.create": {"permission": "refund:create", "mutating": True},
}


def validate_arguments(tool: str, arguments: dict[str, Any]) -> dict[str, Any]:
    if tool == "customer.read":
        if set(arguments) != {"customer_id"}:
            raise ValueError("customer.read requires only customer_id")
        customer_id = arguments["customer_id"]
        if not isinstance(customer_id, str) or not customer_id.startswith("cus_"):
            raise ValueError("customer_id must start with cus_")
        return {"customer_id": customer_id}


    if tool == "refund.create":
        if set(arguments) != {"payment_id", "amount_cents"}:
            raise ValueError("refund.create requires payment_id and amount_cents")
        payment_id = arguments["payment_id"]
        amount = arguments["amount_cents"]
        if not isinstance(payment_id, str) or not payment_id.startswith("pay_"):
            raise ValueError("payment_id must start with pay_")
        if isinstance(amount, bool) or not isinstance(amount, int):
            raise ValueError("amount_cents must be an integer")
        if amount < 1 or amount > 10_000:
            raise ValueError("amount_cents is outside the allowed range")
        return {"payment_id": payment_id, "amount_cents": amount}


    raise ValueError("tool is not registered")


def secure_call(
    principal: Principal,
    tool: str,
    arguments: dict[str, Any],
    approved: bool,
    adapter: Callable[[str, dict[str, Any]], dict[str, Any]],
) -> dict[str, Any]:
    policy = TOOL_POLICY.get(tool)
    if policy is None:
        raise PermissionError("tool is not allowed")
    if policy["permission"] not in principal.permissions:
        raise PermissionError("principal lacks the required permission")
    if policy["mutating"] and not approved:
        raise PermissionError("explicit approval is required")


    validated = validate_arguments(tool, arguments)
    result = adapter(tool, validated)
    if not isinstance(result, dict):
        raise ValueError("tool returned an invalid result")
    return result


def demo_adapter(tool: str, arguments: dict[str, Any]) -> dict[str, Any]:
    return {"tool": tool, "accepted": True, "arguments": arguments}


if __name__ == "__main__":
    principal = Principal(
        user_id="user_42",
        permissions=frozenset({"customer:read", "refund:create"}),
    )
    output = secure_call(
        principal,
        "refund.create",
        {"payment_id": "pay_123", "amount_cents": 2500},
        approved=True,
        adapter=demo_adapter,
    )
    print(output)

入力には、プリンシパル、ツール名、引数オブジェクト、承認判断が含まれます。出力は、アダプターから返される検証済みの辞書です。approved を false に変更すると、実行前に返金リクエストが失敗します。追加の引数、不正な形式の識別子、またはポリシー上限を超える金額を渡した場合も失敗します。本番環境では、権限は信頼できるアイデンティティシステムから取得し、承認はモデルが生成した Boolean 値ではなく、サーバーで検証された証跡として表現する必要があります。

信頼性の高いツール実行

すべてのMCP操作にはタイムアウトが必要です。タイムアウトがないと、利用できないサーバーや下流依存サービスによってワーカーが無期限に占有され、エージェントのワークフローがブロックされる可能性があります。クライアントスタックが対応している場合は、接続用のデッドラインと操作用のデッドラインを分けて設定し、ユーザーリクエスト全体にもデッドラインを設けてください。キャンセルはアプリケーションからMCPクライアントを経由して伝播させ、放棄された処理がリソースを消費し続けないようにする必要があります。

リトライには副作用に関する知識が必要です。リソースの読み取りや決定論的な検索は、一時的なトランスポート障害の後にリトライできることが多くあります。一方、決済、返金、チケット、メッセージの作成は、レスポンスが失われた場合でも最初の試行が成功している可能性があるため、無条件にリトライしてはいけません。状態を変更するツールでは、下流システムが対応している場合、安定した冪等性キーを使用してください。対応していない場合は、再試行する前に操作ステータスを照合してください。

障害は、バリデーション、認可、タイムアウト、トランスポート、サーバー、レート制限、アプリケーションの各エラーに分類する必要があります。ユーザーには、アクションを完了できなかったことを示す安全なメッセージを返し、内部テレメトリには対応可能なカテゴリを記録します。スタックトレース、サーバーアドレス、アクセストークン、下流サービスからの生のレスポンスをモデルに公開してはいけません。サーキットブレーカーは、継続的に障害が発生しているサーバーへの呼び出しを一時的に停止できますが、タイムアウトやモニタリングの代替ではなく、それらを補完するものとして使用してください。

構造化ロギングとトレーシング

オブザーバビリティによって、どの操作が試行されたのか、誰が開始したのか、ポリシーによって許可されたのか、どれだけ時間がかかったのか、なぜ失敗したのかを把握できる必要があります。自由形式のメッセージではなく、構造化イベントを使用してください。役立つフィールドには、タイムスタンプ、トレース識別子、リクエスト識別子、プリンシパル識別子、サーバー名、ツール名、結果、エラーカテゴリ、試行回数、所要時間などがあります。各コンポーネントがトレーシングに対応している場合は、アプリケーション、MCPクライアント、サーバー、下流サービスの間で同じトレースコンテキストを伝播させてください。

デフォルトでは、ツールの引数や結果を完全な形でログに記録しないでください。顧客レコード、認証情報、ドキュメントの内容、モデルのプロンプトが含まれている可能性があります。レコード種別、結果件数、ペイロードサイズなど、安全なメタデータの許可リストを使用することを推奨します。ログ境界でシークレットをマスキングし、アクセス制御と保持期間の制限によってログを保護してください。識別子のハッシュ化は相関分析に役立ちますが、予測可能な識別子は復元される可能性があるため、仮名化が必要な場合はキー付きハッシュのほうが安全です。

メトリクスには、リクエスト量、成功率、拒否された呼び出しの件数、バリデーション失敗、レイテンシのパーセンタイル、タイムアウト、リトライ回数、サーバー可用性を含める必要があります。アラートは、単発のノイズではなくユーザーへの影響を示すものであるべきです。たとえば、重要なツールの失敗率が有意な時間枠にわたって高い状態にある場合にアラートを発生させます。機密性の高いアクションの監査記録は永続化し、隠れたモデル推論を保存することなく、ポリシー判断と承認の根拠を特定できるようにしてください。

信頼性と可観測性を備えたゲートウェイ

次の例では、先ほどの境界にタイムアウト、保守的なリトライ、構造化ロギング、マスキングを追加します。非変更としてマークされたツールのみをリトライします。Pythonのスレッドは基盤で実行中の処理を強制的に停止できないため、アダプター自体でもトランスポートレベルのデッドラインを設定する必要があります。ゲートウェイのタイムアウトによって、呼び出し元のワークフローが無期限に待機することを防ぎ、制御を置くべき場所も示します。

python
import concurrent.futures
import json
import time
import uuid
from typing import Any, Callable


SAFE_TO_RETRY = {"customer.read"}


def emit_event(event: dict[str, Any]) -> None:
    print(json.dumps(event, separators=(",", ":"), sort_keys=True))


def call_with_controls(
    tool: str,
    arguments: dict[str, Any],
    adapter: Callable[[str, dict[str, Any]], dict[str, Any]],
    timeout_seconds: float = 1.0,
    max_attempts: int = 2,
) -> dict[str, Any]:
    request_id = str(uuid.uuid4())
    attempts = max_attempts if tool in SAFE_TO_RETRY else 1


    for attempt in range(1, attempts + 1):
        started = time.monotonic()
        try:
            with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
                future = pool.submit(adapter, tool, arguments)
                result = future.result(timeout=timeout_seconds)


            duration_ms = round((time.monotonic() - started) * 1000, 2)
            emit_event({
                "event": "mcp_tool_call",
                "request_id": request_id,
                "tool": tool,
                "attempt": attempt,
                "outcome": "success",
                "duration_ms": duration_ms,
                "argument_keys": sorted(arguments.keys()),
            })
            return result
        except concurrent.futures.TimeoutError:
            category = "timeout"
        except ConnectionError:
            category = "transport"
        except Exception:
            category = "internal"


        duration_ms = round((time.monotonic() - started) * 1000, 2)
        emit_event({
            "event": "mcp_tool_call",
            "request_id": request_id,
            "tool": tool,
            "attempt": attempt,
            "outcome": "failure",
            "error_category": category,
            "duration_ms": duration_ms,
            "argument_keys": sorted(arguments.keys()),
        })


        if attempt == attempts:
            raise RuntimeError(f"tool call failed: {category}")
        time.sleep(0.1 * attempt)


    raise RuntimeError("unreachable")


def stable_adapter(tool: str, arguments: dict[str, Any]) -> dict[str, Any]:
    return {"customer_id": arguments["customer_id"], "status": "active"}


if __name__ == "__main__":
    result = call_with_controls(
        "customer.read",
        {"customer_id": "cus_123"},
        stable_adapter,
    )
    print(json.dumps(result))

プログラムを実行すると、構造化されたログイベントが1件出力された後に、アダプターの結果が出力されます。ログには引数名が記録されますが値は記録されないため、意図しないデータ漏えいを抑制できます。本番実装では、printによる出力を組織のテレメトリエクスポーターに置き換え、既存のトレース識別子を付加し、元の型付き例外を内部で保持してください。モデル向けのレイヤーには、ユーザーに質問するか、別のツールを選択するか、処理を停止するかを判断するのに役立つ、安定したサニタイズ済みのエラーオブジェクトを返す必要があります。

テストと運用準備

セキュリティテストでは、未知のツール、権限不足、不正な形式の引数、過大な文字列、想定外のフィールド、偽造された承認値、そしてサーバーから返される悪意のあるコンテンツを試行する必要があります。信頼性テストでは、応答の遅延、接続切断、レート制限、部分的な失敗、変更を伴う呼び出しで結果が曖昧になるケースをシミュレートします。リトライ処理によって副作用が重複することが決してないこと、またタイムアウト処理によって完了した可能性のあるトランザクションを確実に失敗したものとして誤って報告しないことを検証してください。

オブザーバビリティのテストも同様に重要です。すべてのリクエストに相関識別子が付与されていること、拒否された呼び出しによって監査イベントが生成されること、レイテンシの測定に単調増加するクロックが使用されていること、機密情報がログに一切出力されないことを確認します。意図的に制御された障害を発生させ、ダッシュボードとアラートにそれらが表示されることを検証してください。一度もテストされていないアラートは、監視パイプラインに関する単なる仮定にすぎません。

したがって、本番環境に対応した MCP 統合では、モデルの意図と実際の機能の間に、制御されたゲートウェイを配置します。このゲートウェイは、入力と出力の両方向を検証し、最小権限ポリシーを適用し、重大な結果を伴うアクションには承認を取得し、期限を設定し、安全な操作に限ってリトライを行い、障害情報をサニタイズし、相関付けられたテレメトリを出力します。これらの制御により、セキュリティルールの適用をモデルに依存せずにシステムの安全性を高められます。また、実際のアプリケーションが開発時のデモンストレーションとは異なる挙動を示した場合に、障害を診断するために必要な証拠も提供できます。

レッスンの練習

レッスンのチェックポイント

学んだ内容を復習し、回答へのフィードバックを受けましょう。

0/1
回答済み · 0/1