コンテンツにスキップ

第8章 ツールを束ねる「ゲートウェイ」 — 学習メモ

書籍「Amazon Bedrock AgentCore実践入門」第8章のサンプルコード(このフォルダ)を読み解いた個人学習メモ。 AgentCore 全体像は ../../lectures/agentcore_basics/STUDY_NOTES.md 参照。 実行検証は伴わない。コードに現れた API 名だけ断定し、読めない挙動は「〜と推測」で明示する。

一言で

ゲートウェイ = 複数の異なるツール(既存の MCP サーバーや、別の AgentCore ランタイム)を1つの HTTPS エンドポイント+IAM認証にまとめ、間にインターセプター(Lambda)を挟んで検査・改変・遮断できるようにする「ツールの集約レイヤー」。エージェント側から見ると、バラバラなツール群が単一の接続先に見える。

全体像

3ファイル構成。ゲートウェイを間に挟むことで、エージェント(呼び出し元)とツール実体(MCPサーバー・別ランタイム)が疎結合になり、間で検査を挟める。

ファイル 役割
01_mcp_client.py ゲートウェイに MCP 接続し、ツールとしてエージェントに渡す
02_http_target.py boto3 で直接ゲートウェイの HTTP ターゲットを叩き、別の AgentCore ランタイムを呼び出す
03_interceptor.py ゲートウェイのインターセプターとして Lambda にデプロイするコード。SQL インジェクションパターンを検査して遮断/通過を判定する(ローカル実行しない)
pyproject.toml strands-agents==1.38.0 を固定。mcp-proxy-for-aws はバージョン非固定(このリポジトリの他章は == 固定が方針なのでこれは例外)
flowchart TD
    Ag["Strandsエージェント<br>(01_mcp_client.py)"]
    GW["AgentCoreゲートウェイ<br>(IAM認証で集約)"]
    Int["インターセプターLambda<br>(03_interceptor.py)"]
    MCPT["MCPターゲット<br>(既存MCPサーバー)"]
    HTTPT["HTTPターゲット<br>(別のAgentCoreランタイム, 02_http_target.py)"]
    Block["403 Forbiddenで遮断<br>(SQLパターン検出時)"]

    Ag -->|"aws_iam_streamablehttp_client"| GW
    GW --> Int
    Int -->|"tools/call かつ問題なし"| MCPT
    Int -->|"tools/call かつ問題なし"| HTTPT
    Int -.->|"SQL_PATTERNS検出"| Block

使用ライブラリ・原理

  • strands.tools.mcp.MCPClientstrands-agents==1.38.0): MCP 接続をラップして Strands の tools にそのまま渡せるクライアント
  • mcp_proxy_for_aws.client.aws_iam_streamablehttp_client: ゲートウェイの MCP エンドポイントに AWS SigV4(IAM)署名付きで streamable HTTP 接続するクライアント
  • boto3bedrock-agentcore サービスクライアント: invoke_agent_runtime で HTTP ターゲット経由の呼び出しを行う(Strands を介さない素の呼び出し)
  • Lambda ハンドラー(json, re のみ): ゲートウェイのインターセプター契約に沿った検査ロジック

中心コードの読み解き

01_mcp_client.py(ゲートウェイへの MCP 接続)

mcp_client = MCPClient(lambda: aws_iam_streamablehttp_client(
    endpoint="https://<ゲートウェイID>.gateway.bedrock-agentcore.us-east-1.amazonaws.com/mcp",
    aws_service="bedrock-agentcore"
))

agent = Agent(tools=[mcp_client])
agent("注文情報を検索してください")
やってること なぜ
01_mcp_client.py:6-9 aws_iam_streamablehttp_client でゲートウェイの MCP エンドポイントに IAM 署名接続 ゲートウェイが IAM 認証を要求するため、素の HTTP クライアントではなく AWS SigV4 署名するクライアントが必要
01_mcp_client.py:12 Agent(tools=[mcp_client]) MCPClient インスタンスをそのまま tools リストに渡せる。ゲートウェイ配下の全ツールが1本の MCP 接続で使えるようになる

02_http_target.py(HTTP ターゲット経由で別ランタイムを呼ぶ)

client = boto3.client(
    service_name="bedrock-agentcore",
    endpoint_url="https://<ゲートウェイID>.gateway.bedrock-agentcore.us-east-1.amazonaws.com/<ターゲット名>"
)

response = client.invoke_agent_runtime(
   agentRuntimeArn="arn:aws:bedrock-agentcore:us-east-1:xxxxxxxxxxxx:runtime/handson_MyAgent-XXXXXXXXXX",
   runtimeSessionId="this-is-runtime-session-id-000001",
   payload=json.dumps({"prompt": "こんにちは"})
)
やってること なぜ
02_http_target.py:5-8 boto3.client(..., endpoint_url=".../<ターゲット名>") ゲートウェイの HTTP ターゲット URL を endpoint_url に直接指定し、通常の bedrock-agentcore クライアントとして扱う
02_http_target.py:11-15 client.invoke_agent_runtime(agentRuntimeArn=..., payload=...) ゲートウェイの HTTP ターゲットを経由して、その先にある別の AgentCore ランタイムを呼び出す。Strands の Agent を介さない素の boto3 呼び出しである点が 01_mcp_client.py と対照的

03_interceptor.py(Lambda インターセプター)

def lambda_handler(event, context):
    mcp_data = event.get("mcp", {})
    request = mcp_data.get("gatewayRequest", {})
    body = request.get("body", {})

    if body.get("method") != "tools/call":
        return {
            "interceptorOutputVersion": "1.0",
            "mcp": {"transformedGatewayRequest": request}
        }

    params_str = json.dumps(body.get("params", {}), ensure_ascii=False)
    for pattern in SQL_PATTERNS:
        if re.search(pattern, params_str, re.IGNORECASE):
            return {
                "interceptorOutputVersion": "1.0",
                "mcp": {"transformedGatewayResponse": {"statusCode": 403, "body": body}}
            }

    return {
        "interceptorOutputVersion": "1.0",
        "mcp": {"transformedGatewayRequest": request}
    }
やってること なぜ
03_interceptor.py:12-14 event["mcp"]["gatewayRequest"]["body"] を取り出す ゲートウェイが Lambda に渡すイベント形式(mcp.gatewayRequest.body)に沿ってリクエスト本体を取得する
03_interceptor.py:17-24 method != "tools/call" ならそのまま透過 ツール呼び出し以外(ツール一覧取得等)は検査対象外として素通しし、オーバーヘッドを避ける
03_interceptor.py:27-34 SQL_PATTERNS の正規表現で params を検査 ツールの実引数に SQL インジェクションらしき文字列がないかを検査する
03_interceptor.py:36-44 検出時は transformedGatewayResponsestatusCode: 403 を返す ツール本体を一切呼ばせずゲートウェイ側で遮断する。これは「モデルの出力を信じない」構造の防御
03_interceptor.py:47-51 問題なければ transformedGatewayRequest で元の request を返す 素通しして実際のツール呼び出しへ進める

学んだこと(要点)

  • ゲートウェイは「エンドポイントの統一」と「IAM 認証」を担う集約装置。ツールの実体(MCP サーバーや別ランタイム)はゲートウェイの背後に隠れ、エージェントからは1つの接続先にしか見えない
  • インターセプターの応答契約は2択: {"interceptorOutputVersion": "1.0", "mcp": {"transformedGatewayRequest": ...}}(通過)または {"transformedGatewayResponse": {"statusCode": ..., "body": ...}}(遮断)
  • SQL_PATTERNS は3パターン(;+DML系、UNION SELECTDROP TABLE)のみのデモであり、正規表現ベースなので迂回の余地がある
  • 第9章の Cedar ポリシー(Policy)とは別レイヤの防御。インターセプターは「任意のロジックで動的検査」、Cedar は「宣言的な permit/forbid」。どちらもゲートウェイに刺さる別の防御点であり、両方を重ねて使える

落とし穴・現代版に移植するなら

  • <ゲートウェイID> <ターゲット名> は README の指示通りプレースホルダーであり、置換必須(そのままでは動かない)
  • mcp-proxy-for-awspyproject.toml でバージョン非固定(このリポジトリの他章は == 固定が方針なので例外的な扱い)
  • 03_interceptor.py は Lambda ハンドラーなので uv run でのローカル実行はできない(デプロイ専用のコード)
  • SQL_PATTERNS は正規表現ベースで迂回可能なため、実運用ではパラメータ化クエリ側(ツール実装側)の対策と併用すべき、という点はコードから読める事実ではなく一般的な補足

記事参照


作成: 2026-07-17 / 最終更新: 2026-07-17