AgentCore RuntimeからAgentCore GatewayへのIAM認証をMCP Proxy for AWSで実装してみた

import asyncio
from bedrock_agentcore.identity.auth import requires_access_token, requires_api_key

@requires_access_token(
    provider_name= "sample-provider",
    scopes= [],
    auth_flow= 'M2M',
)
async def need_token_2LO_async(*, access_token: str):

resource "aws_bedrockagentcore_gateway" "main" {
  name     = "${var.project_name}-gateway"
  role_arn = aws_iam_role.gateway.arn

  protocol_type = "MCP"

  # Inbound認証: IAM認証を指定
  authorizer_type = "AWS_IAM"

  # ...省略
}

resource "aws_iam_role_policy" "agent_runtime_gateway" {
  name = "${var.project_name}-agent-runtime-gateway-policy"
  role = aws_iam_role.agent_runtime.id

  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Sid    = "InvokeGateway"
        Effect = "Allow"
        Action = [
          "bedrock-agentcore:InvokeGateway"
        ]
        Resource = aws_bedrockagentcore_gateway.main.gateway_arn
      }
    ]
  })
}

{
  "Action": "bedrock-agentcore:InvokeGateway",
  "Resource": "arn:aws:bedrock-agentcore:us-east-1:123456789012:gateway/my-gateway"
}

import os
from bedrock_agentcore.runtime import BedrockAgentCoreApp
from strands import Agent
from strands.models import BedrockModel
from strands.tools.mcp import MCPClient
from mcp_proxy_for_aws.client import aws_iam_streamablehttp_client

GATEWAY_URL = os.getenv("GATEWAY_URL", "")
AWS_REGION = os.getenv("AWS_REGION", "us-east-1")
MODEL_ID = os.getenv("BEDROCK_MODEL_ID", "us.anthropic.claude-haiku-4-5-20251001-v1:0")

app = BedrockAgentCoreApp()
model = BedrockModel(model_id=MODEL_ID, region_name=AWS_REGION)

@app.entrypoint
def invoke(payload: dict) -> dict:
    message = payload.get("prompt") or payload.get("message", "")
    if not message:
        return {"error": "No message", "status": "error"}

    if GATEWAY_URL:
        # mcp-proxy-for-aws で SigV4 署名付きの MCP クライアントを作成
        mcp_factory = lambda: aws_iam_streamablehttp_client(
            endpoint=GATEWAY_URL,
            aws_region=AWS_REGION,
            aws_service="bedrock-agentcore"
        )
        mcp_client = MCPClient(mcp_factory)
        agent = Agent(model=model, tools=[mcp_client])
        result = agent(message)
    else:
        agent = Agent(model=model)
        result = agent(message)

    return {"response": str(result.message), "status": "success"}

if __name__ == "__main__":
    app.run()

mcp_factory = lambda: aws_iam_streamablehttp_client(
    endpoint=GATEWAY_URL,
    aws_region=AWS_REGION,
    aws_service="bedrock-agentcore"
)

mcp_factory = lambda: aws_iam_streamablehttp_client(
            endpoint=GATEWAY_URL,
            aws_region=AWS_REGION,
            aws_service="bedrock-agentcore"
        )
with MCPClient(mcp_factory) as mcp:
    agent = Agent(model=model, tools=mcp.list_tools_sync())
    result = agent(message)

git clone https://github.com/yuu551/runtime-to-gateway-for-iam.git
cd runtime-to-gateway-for-iam

terraform init

terraform plan

terraform apply

Do you want to perform these actions?
  Terraform will perform the actions described above.
  Only 'yes' will be accepted to approve.

  Enter a value: yes

Apply complete! Resources: 23 added, 0 changed, 0 destroyed.

Outputs:

gateway_endpoint = "https://xxxxxxxxxx.bedrock-agentcore.us-east-1.amazonaws.com"
runtime_endpoint = "https://yyyyyyyyyy.bedrock-agentcore.us-east-1.amazonaws.com"

{"prompt": "ツールで何ができるか教えて"}

{"prompt": "100 + 12は"}

{
  "message": {
    "tool_calls": [
      {
        "type": "function",
        "id": "tooluse_c9qOEPXPS2a6OR8UxN4tfg",
        "function": {
          "name": "iam-auth-demo-lambda-target___add_numbers",
          "arguments": {
            "a": 100,
            "b": 12
          }
        }
      }
    ],
    "role": "assistant"
  },
  "index": 0,
  "finish_reason": "tool_use"
}