
【小ネタ】AgentCore Gatewayのインターセプターにカスタムヘッダーを渡してみた
はじめに
こんにちは、ドライブがうまくなりたいコンサルティング部の神野です。(買ったばかりの車を傷つけて泣いています)
Amazon Bedrock AgentCore Gatewayにはインターセプターという仕組みがあり、ツール呼び出しの前後でLambda関数による検証や変換を挟むことができます。以前にもリクエスト・レスポンスインターセプターそれぞれブログを書きました。
この仕組みを利用検討していて、MCPクライアントが付けたカスタムHTTPヘッダーはインターセプターに届くのか?が気になりました。
呼び出し元のアプリ識別子などの固有な情報をヘッダーで渡してインターセプター側で使う、といった設計ができるかどうかに関わるポイントですよね。実際にGatewayとCognitoで環境を作って確かめてみました!
いきなり結論
カスタムヘッダーはインターセプターに送信することは可能です。ただし条件があります。
- インターセプター設定で passRequestHeaders を true にした場合のみ、リクエストヘッダーが gatewayRequest.headers に渡される
- passRequestHeaders 未設定の場合、今回の検証では headers は空のマップ
{}になり、カスタムヘッダーは一切渡らなかった - 今回付与したカスタムヘッダーは小文字化されて届いた(X-Dept-Id → x-dept-id)。参照時は小文字化して扱うのが安全
- passRequestHeaders を有効にすると Authorization ヘッダー(Bearerトークン)もそのまま渡るため、ログ出力には注意が必要
なお、届くことと制御に使ってよいことは別問題です。カスタムヘッダーはクライアントが自由に書き換えられる値なので、認可などの制御に使うことは推奨しません。経路上でヘッダーの改ざんがないことが担保されている(信頼できるプロキシが付与・上書きする等の)構成である場合に限り、選択肢になり得ます。この点は記事の後半でも触れます。
次のセクションから、検証環境の構築と実際の挙動を確認していきます。
前提
公式ドキュメントには、インターセプターのセキュリティベストプラクティスとして下記の記載があります。
By default, request headers will not be passed to an interceptor unless the
passRequestHeadersfield is set to true. Be careful when using this field as request headers can contain sensitive information such as authentication tokens and credentials.
つまりヘッダーはデフォルトでは渡されず、明示的にオプトインする設計です。ドキュメント上の仕様はわかったので、実際にカスタムヘッダーがどんな形で届くのかを確認していきます。
検証環境は下記の通りです。
- リージョン: us-east-1
- boto3 / botocore: 1.43.43
- インバウンド認証: Amazon Cognito User Pool(client_credentialsフロー)
- インターセプター: Python 3.13 の Lambda 関数(REQUESTインターセプター)
Cognito User Pool・リソースサーバー・アプリクライアントの作成手順は本題ではないので折りたたんでおきます。
Cognitoリソースの作成コマンド
# User Pool
aws cognito-idp create-user-pool \
--pool-name header-demo-pool \
--query 'UserPool.Id' --output text
# => us-east-1_xxxxxxxxx
# リソースサーバー
aws cognito-idp create-resource-server \
--user-pool-id us-east-1_xxxxxxxxx \
--identifier gateway-api \
--name "Gateway API" \
--scopes ScopeName=invoke,ScopeDescription="Invoke gateway tools"
# トークンエンドポイント用ドメイン
aws cognito-idp create-user-pool-domain \
--domain header-demo-xxxxxxxx \
--user-pool-id us-east-1_xxxxxxxxx
# client_credentialsフローのアプリクライアント
aws cognito-idp create-user-pool-client \
--user-pool-id us-east-1_xxxxxxxxx \
--client-name gateway-m2m-client \
--generate-secret \
--allowed-o-auth-flows client_credentials \
--allowed-o-auth-scopes "gateway-api/invoke" \
--allowed-o-auth-flows-user-pool-client
構築
インターセプターLambdaの作成
まず、届いたイベントをそのままログに出力してパススルーするだけのインターセプターLambdaを作成します。何が届くのかを観察するのが目的です。
import json
def lambda_handler(event, context):
print("INTERCEPTOR_EVENT " + json.dumps(event, ensure_ascii=False))
body = event.get("mcp", {}).get("gatewayRequest", {}).get("body", {})
return {
"interceptorOutputVersion": "1.0",
"mcp": {
"transformedGatewayRequest": {"body": body}
},
}
REQUESTインターセプターは interceptorOutputVersion を "1.0" として、mcp.transformedGatewayRequest に転送するリクエストボディを入れて返す必要があります。ここでは受け取ったボディをそのまま返すことでパススルーしています。入出力ペイロードの形式は下記ドキュメントに詳しくまとまっています。
コードができたら、実行ロールを作ってLambda関数をデプロイします。
# Lambda実行ロールの作成
cat > lambda-trust.json <<'EOF'
{
"Version": "2012-10-17",
"Statement": [{
"Effect": "Allow",
"Principal": {"Service": "lambda.amazonaws.com"},
"Action": "sts:AssumeRole"
}]
}
EOF
aws iam create-role \
--role-name header-demo-lambda-role \
--assume-role-policy-document file://lambda-trust.json
aws iam attach-role-policy \
--role-name header-demo-lambda-role \
--policy-arn arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole
# zip化してデプロイ
cd interceptor && zip ../interceptor.zip lambda_function.py && cd ..
aws lambda create-function \
--function-name header-demo-interceptor \
--runtime python3.13 \
--role arn:aws:iam::123456789012:role/header-demo-lambda-role \
--handler lambda_function.lambda_handler \
--zip-file fileb://interceptor.zip \
--timeout 10
実行ロールには AWSLambdaBasicExecutionRole のみアタッチしています。イベントの中身はCloudWatch Logsで確認するので、ログ出力の権限があれば十分です。
Gateway実行ロールの作成
Gatewayがインターセプターとターゲットの各Lambdaを呼び出せるよう、実行ロールを作成します。
cat > gateway-trust.json <<'EOF'
{
"Version": "2012-10-17",
"Statement": [{
"Effect": "Allow",
"Principal": {"Service": "bedrock-agentcore.amazonaws.com"},
"Action": "sts:AssumeRole",
"Condition": {"StringEquals": {"aws:SourceAccount": "123456789012"}}
}]
}
EOF
aws iam create-role \
--role-name header-demo-gateway-role \
--assume-role-policy-document file://gateway-trust.json
cat > gateway-policy.json <<'EOF'
{
"Version": "2012-10-17",
"Statement": [{
"Effect": "Allow",
"Action": ["lambda:InvokeFunction"],
"Resource": [
"arn:aws:lambda:us-east-1:123456789012:function:header-demo-interceptor",
"arn:aws:lambda:us-east-1:123456789012:function:header-demo-tool"
]
}]
}
EOF
aws iam put-role-policy \
--role-name header-demo-gateway-role \
--policy-name lambda-invoke \
--policy-document file://gateway-policy.json
公式のベストプラクティスに従いインターセプターとツールのLambdaを実行できる権限に絞っています。
Gatewayの作成
インターセプター設定付きでGatewayを作成します。ここが今回のポイントで、inputConfiguration の passRequestHeaders を true にします。
import boto3
client = boto3.client("bedrock-agentcore-control", region_name="us-east-1")
response = client.create_gateway(
name="header-demo",
roleArn="arn:aws:iam::123456789012:role/header-demo-gateway-role",
protocolType="MCP",
authorizerType="CUSTOM_JWT",
authorizerConfiguration={
"customJWTAuthorizer": {
"discoveryUrl": "https://cognito-idp.us-east-1.amazonaws.com/us-east-1_xxxxxxxxx/.well-known/openid-configuration",
"allowedClients": ["<CognitoアプリクライアントID>"],
}
},
interceptorConfigurations=[
{
"interceptor": {
"lambda": {
"arn": "arn:aws:lambda:us-east-1:123456789012:function:header-demo-interceptor"
}
},
"interceptionPoints": ["REQUEST"],
"inputConfiguration": {"passRequestHeaders": True},
}
],
)
print(response["gatewayUrl"])
インターセプター設定の項目は下記の通りです。
| 設定項目 | 設定値 | 説明 |
|---|---|---|
| interceptor.lambda.arn | LambdaのARN | インターセプターとして呼び出すLambda関数 |
| interceptionPoints | REQUEST | ターゲット呼び出し前に実行(RESPONSEも指定可) |
| inputConfiguration.passRequestHeaders | True | リクエストヘッダーをインターセプターに渡す |
roleArn には先ほど作成したGateway実行ロールを指定します。インターセプターは1つのGatewayにつきREQUEST・RESPONSE各1つまでという制約があります。
動作確認用のツールとして、挨拶を返すだけのLambda関数をターゲット登録しておきます。
ツールLambdaとターゲット登録
def lambda_handler(event, context):
name = event.get('name', 'world')
return {'greeting': f'Hello, {name}!'}
zip tool.zip lambda_function.py
aws lambda create-function \
--function-name header-demo-tool \
--runtime python3.13 \
--role arn:aws:iam::123456789012:role/header-demo-lambda-role \
--handler lambda_function.lambda_handler \
--zip-file fileb://tool.zip
client.create_gateway_target(
gatewayIdentifier="<Gateway ID>",
name="greeting-tool",
targetConfiguration={
"mcp": {
"lambda": {
"lambdaArn": "arn:aws:lambda:us-east-1:123456789012:function:header-demo-tool",
"toolSchema": {
"inlinePayload": [
{
"name": "greet",
"description": "Return a greeting for the given name",
"inputSchema": {
"type": "object",
"properties": {"name": {"type": "string"}},
"required": ["name"],
},
}
]
},
}
}
},
credentialProviderConfigurations=[{"credentialProviderType": "GATEWAY_IAM_ROLE"}],
)
これで準備完了です。実際に試してみます!
動作確認
カスタムヘッダー付きでツールを呼び出す
Cognitoから取得したトークンに加えて、カスタムヘッダーを3つ付けてツールを呼び出します。
curl -s -X POST https://<Gateway ID>.gateway.bedrock-agentcore.us-east-1.amazonaws.com/mcp \
-H "Authorization: Bearer $TOKEN" \
-H 'Content-Type: application/json' \
-H 'Accept: application/json, text/event-stream' \
-H 'X-Dept-Id: dept-a' \
-H 'X-App-Id: hr-agent-001' \
-H 'X-Trace-Note: custom-header-demo' \
-d '{"jsonrpc":"2.0","id":10,"method":"tools/call","params":{"name":"greeting-tool___greet","arguments":{"name":"Interceptor"}}}'
{
"jsonrpc": "2.0",
"id": 10,
"result": {
"isError": false,
"content": [
{"type": "text", "text": "{\"greeting\":\"Hello, Interceptor!\"}"}
]
}
}
呼び出しはインターセプター経由で問題なく成功していますね。それでは本題、インターセプターLambdaに届いたイベントをCloudWatch Logsで確認します。
{
"interceptorInputVersion": "1.0",
"mcp": {
"gatewayRequest": {
"path": "/mcp",
"httpMethod": "POST",
"headers": {
"Host": "<Gateway ID>.gateway.bedrock-agentcore.us-east-1.amazonaws.com",
"authorization": "Bearer eyJraWQ...",
"accept": "application/json, text/event-stream",
"content-type": "application/json",
"user-agent": "curl/8.7.1",
"x-dept-id": "dept-a",
"x-app-id": "hr-agent-001",
"x-trace-note": "custom-header-demo",
"X-Forwarded-For": "203.0.113.10",
"X-Amzn-Trace-Id": "Root=1-...",
"x-amzn-tls-version": "TLSv1.3"
},
"body": {
"id": 10,
"jsonrpc": "2.0",
"method": "tools/call",
"params": {
"name": "greeting-tool___greet",
"arguments": {"name": "Interceptor"}
}
}
},
"rawGatewayRequest": {
"body": "{\"jsonrpc\":\"2.0\",\"id\":10,...}"
}
}
}
お、カスタムヘッダー3つがすべて届いていますね!今回付与したカスタムヘッダーは x-dept-id / x-app-id / x-trace-note と小文字に正規化されて届いています。一方で Host や X-Forwarded-For のように大文字を含むヘッダーも混在しているので、インターセプター側で参照する際はキーを小文字化してから扱うのが安全です。加えて、クライアントの送信元IP(X-Forwarded-For)やUser-Agentといったメタ情報も一緒に取れることがわかります。
passRequestHeadersなしの場合と比較する
比較のため、update-gatewayで inputConfiguration を外した(passRequestHeaders未設定の)状態でも同じ呼び出しをしてみました。
{
"interceptorInputVersion": "1.0",
"mcp": {
"gatewayRequest": {
"path": "/mcp",
"httpMethod": "POST",
"headers": {},
"body": {"id": 11, "jsonrpc": "2.0", "method": "tools/call", ...}
}
}
}
headers が空のマップになりました。カスタムヘッダーはもちろん、Authorizationヘッダーも一切付与されていませんね。ヘッダーを制御に使いたい場合は passRequestHeaders の明示的な有効化が必須ということが確認できましたね!
届いたヘッダーで制御してみる
ヘッダーが取れることがわかったので、ヘッダー値による制御が技術的に成立するかも挙動確認として試してみます。X-Dept-Id が許可リストにない場合、ターゲットを呼ばずにインターセプターがエラーを短絡応答する実装です。後述の通り、この方式を認可として本番採用することは推奨しません。あくまで動くかどうかの確認です。
import json
ALLOWED_DEPTS = {"dept-a"}
def lambda_handler(event, context):
req = event.get("mcp", {}).get("gatewayRequest", {})
headers = {k.lower(): v for k, v in (req.get("headers") or {}).items()}
body = req.get("body", {})
dept = headers.get("x-dept-id")
if body.get("method") == "tools/call" and dept not in ALLOWED_DEPTS:
return {
"interceptorOutputVersion": "1.0",
"mcp": {
"transformedGatewayResponse": {
"statusCode": 403,
"body": {
"jsonrpc": "2.0",
"id": body.get("id"),
"error": {
"code": -32000,
"message": f"Access denied: department '{dept}' is not allowed to call tools",
},
},
}
},
}
return {
"interceptorOutputVersion": "1.0",
"mcp": {"transformedGatewayRequest": {"body": body}},
}
transformedGatewayResponse を返すと、Gatewayはターゲットを呼び出さずその内容を即座にクライアントへ返します。X-Dept-Id: dept-b で呼び出した結果が下記です。
{"jsonrpc":"2.0","id":21,"error":{"code":-32000,"message":"Access denied: department 'dept-b' is not allowed to call tools"}}
HTTP_STATUS=403
HTTP 403とJSON-RPCエラーで拒否されました!dept-a を付けた呼び出しは従来どおり成功するので、ヘッダーベースの制御が機能していることが確認できましたね。
おわりに
passRequestHeaders ひとつでヘッダーの見え方が完全に切り替わる、明快なオプトイン設計でした。カスタムヘッダーを使う場合は、passRequestHeaders の明示的な有効化とログのマスクをセットで検討したいですね。インターセプター自体はある程度柔軟に実装できて認可・変換・監査と使い道が広いので、必要になるシーンで引き続き活用していきたいと思います!
本記事が少しでも参考になりましたら幸いです。最後までご覧いただきありがとうございました!
補足: リクエストボディの参照・加工もできる
REQUESTインターセプターはリクエストボディの参照や加工もできます。イベントの gatewayRequest.body にはJSON-RPCリクエストがパース済みのJSONで入ってくるので、メソッド・ツール名・引数の中身まで取り出して判断に使えます(本文のデモでも body の method を見て tools/call だけを判定対象にしていました)。判断した結果は、そのまま通す・書き換えて通す・transformedGatewayResponse で短絡応答する、と分岐できます。書き換える場合は transformedGatewayRequest.body に入れて返した内容がそのままターゲットへ転送されます。
試しに、tools/call の arguments.name を書き換えるインターセプターに差し替えてみます。
import json
def lambda_handler(event, context):
body = event.get("mcp", {}).get("gatewayRequest", {}).get("body", {})
if body.get("method") == "tools/call":
args = body.get("params", {}).get("arguments", {})
if "name" in args:
args["name"] = args["name"] + " (rewritten by interceptor)"
return {
"interceptorOutputVersion": "1.0",
"mcp": {"transformedGatewayRequest": {"body": body}},
}
クライアントからは name を Jinno として呼び出します。
{
"jsonrpc": "2.0",
"id": 30,
"result": {
"isError": false,
"content": [
{"type": "text", "text": "{\"greeting\":\"Hello, Jinno (rewritten by interceptor)!\"}"}
]
}
}
クライアントが送った値ではなく、インターセプターで書き換えた値がターゲットのLambdaに届いていますね!引数へのコンテキスト注入や特定フィールドのマスクといった用途に使えます。
なお、httpMethod と path は読み取り専用で変更できません。また、HTTPターゲット(AgentCore Runtime等)の場合はボディがbase64エンコード文字列で渡るため、デコードしてから加工する必要があります。








