第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.MCPClient(strands-agents==1.38.0): MCP 接続をラップして Strands のtoolsにそのまま渡せるクライアントmcp_proxy_for_aws.client.aws_iam_streamablehttp_client: ゲートウェイの MCP エンドポイントに AWS SigV4(IAM)署名付きで streamable HTTP 接続するクライアントboto3のbedrock-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 |
検出時は transformedGatewayResponse で statusCode: 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 SELECT、DROP TABLE)のみのデモであり、正規表現ベースなので迂回の余地がある- 第9章の Cedar ポリシー(Policy)とは別レイヤの防御。インターセプターは「任意のロジックで動的検査」、Cedar は「宣言的な permit/forbid」。どちらもゲートウェイに刺さる別の防御点であり、両方を重ねて使える
落とし穴・現代版に移植するなら¶
<ゲートウェイID><ターゲット名>は README の指示通りプレースホルダーであり、置換必須(そのままでは動かない)mcp-proxy-for-awsはpyproject.tomlでバージョン非固定(このリポジトリの他章は==固定が方針なので例外的な扱い)03_interceptor.pyは Lambda ハンドラーなのでuv runでのローカル実行はできない(デプロイ専用のコード)SQL_PATTERNSは正規表現ベースで迂回可能なため、実運用ではパラメータ化クエリ側(ツール実装側)の対策と併用すべき、という点はコードから読める事実ではなく一般的な補足
記事参照¶
- 書籍 第8章。関連:
../../lectures/agentcore_basics/STUDY_NOTES.md(AgentCore 全体像。Gateway は8部品の1つ)
作成: 2026-07-17 / 最終更新: 2026-07-17