[小ネタ] Strands AgentsからAgentCore Gatewayの推論ターゲットを呼び出してみた

import boto3
import httpx
from botocore.auth import SigV4Auth
from botocore.awsrequest import AWSRequest

class GatewaySigV4Auth(httpx.Auth):
    """Gatewayへのリクエストに SigV4 署名を付ける httpx の Auth"""

    requires_request_body = True

    def __init__(self, region: str = "us-east-1"):
        self.region = region
        self.session = boto3.Session()

    def auth_flow(self, request: httpx.Request):
        # SDK が付ける API キー用のヘッダーは Gateway では使わないので外す
        for name in ("authorization", "x-api-key"):
            request.headers.pop(name, None)

        aws_request = AWSRequest(
            method=request.method,
            url=str(request.url),
            data=request.content,
            headers={"content-type": request.headers.get("content-type", "application/json")},
        )
        credentials = self.session.get_credentials().get_frozen_credentials()
        SigV4Auth(credentials, "bedrock-agentcore", self.region).add_auth(aws_request)
        request.headers.update(dict(aws_request.headers))
        yield request

class KeepOpenAsyncClient(httpx.AsyncClient):
    """SDK のクライアントがクローズされても、HTTP クライアントはクローズせずに使い続ける"""

    async def aclose(self) -> None:
        # SDK がクローズしようとしても何もしない
        pass

    async def close_for_real(self) -> None:
        # 使い終わったら自分で呼んでクローズする
        await super().aclose()

def signed_http_client() -> KeepOpenAsyncClient:
    return KeepOpenAsyncClient(auth=GatewaySigV4Auth())

from strands import tool

@tool
def get_weather(city: str) -> str:
    """指定した都市の天気を返します"""
    return f"{city}は晴れ、気温は18度です"

import os

from strands import Agent
from strands.models.anthropic import AnthropicModel

from sigv4_auth import signed_http_client
from tools import get_weather

GATEWAY_URL = os.environ["GATEWAY_URL"]

model = AnthropicModel(
    client_args={
        "base_url": f"{GATEWAY_URL}/inference",
        "api_key": "unused",
        "http_client": signed_http_client(),
    },
    model_id="us.anthropic.claude-haiku-5-5",
    max_tokens=1024,
)

agent = Agent(model=model, tools=[get_weather])
agent("東京と大阪の天気を教えて")

uv run agent_claude.py

Tool #1: get_weather

Tool #2: get_weather
東京と大阪の天気は、どちらも晴れで、気温は18度です。

import asyncio
import os

from strands import Agent
from strands.models.openai_responses import OpenAIResponsesModel

from sigv4_auth import signed_http_client
from tools import get_weather

GATEWAY_URL = os.environ["GATEWAY_URL"]

async def main():
    http_client = signed_http_client()
    model = OpenAIResponsesModel(
        client_args={
            "base_url": f"{GATEWAY_URL}/inference/v1",
            "api_key": "unused",
            "http_client": http_client,
        },
        model_id="openai.gpt-5.6-luna",
    )
    agent = Agent(model=model, tools=[get_weather])
    try:
        await agent.invoke_async("東京と大阪の天気を教えて")
    finally:
        # エージェントを使い終わったら HTTP クライアントをクローズする
        await http_client.close_for_real()

asyncio.run(main())

uv run agent_gpt.py

Tool #1: get_weather

Tool #2: get_weather
- 東京：晴れ、18℃
- 大阪：晴れ、18℃

Tool #1: get_weather
Tool #2: get_weather
    raise RuntimeError("Cannot send a request, as the client has been closed.")
RuntimeError: Cannot send a request, as the client has been closed.

async with openai.AsyncOpenAI(**self._resolve_client_args()) as client:
    try:
        response = await client.responses.create(**request)

import os

from strands.models.anthropic import AnthropicModel
from strands.models.openai_responses import OpenAIResponsesModel

from sigv4_auth import signed_http_client

GATEWAY_URL = os.environ["GATEWAY_URL"]

def claude_model(model_id: str = "chat-claude") -> AnthropicModel:
    """Anthropic Messages 形式で呼ぶモデル"""
    return AnthropicModel(
        client_args={"base_url": f"{GATEWAY_URL}/inference", "api_key": "unused", "http_client": signed_http_client()},
        model_id=model_id,
        max_tokens=1024,
    )

def gpt_model(model_id: str = "chat-gpt") -> OpenAIResponsesModel:
    """OpenAI Responses 形式で呼ぶモデル"""
    return OpenAIResponsesModel(
        client_args={"base_url": f"{GATEWAY_URL}/inference/v1", "api_key": "unused", "http_client": signed_http_client()},
        model_id=model_id,
    )

import os
import sys
import time

import boto3

client = boto3.client("bedrock-agentcore-control", region_name="us-east-1")
gateway_id = os.environ["GATEWAY_URL"].removeprefix("https://").split(".")[0]
timeout_ms = int(sys.argv[1]) if len(sys.argv) > 1 else 30000

# 論理名ごとに、プライマリとフォールバック先（同じ形式で呼べるモデル）を決める
RULES = {
    "chat-claude": (
        {"targetName": "bedrock-runtime", "model": "us.anthropic.claude-haiku-5-5"},
        {"targetName": "bedrock-mantle", "model": "anthropic.claude-haiku-4-5"},
    ),
    "chat-gpt": (
        {"targetName": "bedrock-mantle", "model": "openai.gpt-5.6-luna"},
        {"targetName": "bedrock-runtime", "model": "us.openai.gpt-5.6-luna"},
    ),
}

existing = {
    r["conditions"][0]["matchModels"]["anyOf"][0]["value"]: r["ruleId"]
    for r in client.list_gateway_rules(gatewayIdentifier=gateway_id)["gatewayRules"]
}

for priority, (name, (primary, secondary)) in enumerate(RULES.items(), start=10):
    params = dict(
        gatewayIdentifier=gateway_id,
        priority=priority,
        conditions=[{"matchModels": {"anyOf": [{"value": name, "operator": "StringEquals"}]}}],
        actions=[{
            "routeToTarget": {
                "staticRoute": {
                    **primary,
                    "fallback": {"firstByteTimeoutMilliseconds": timeout_ms, "targetFallback": [secondary]},
                }
            }
        }],
    )
    if name in existing:
        rule_id = existing[name]
        client.update_gateway_rule(ruleId=rule_id, **params)
    else:
        rule_id = client.create_gateway_rule(**params)["ruleId"]
    while client.get_gateway_rule(gatewayIdentifier=gateway_id, ruleId=rule_id)["status"] != "ACTIVE":
        time.sleep(3)
    print(name, "ACTIVE", timeout_ms)

uv run setup_rules.py && uv run check_rules.py
uv run setup_rules.py 50 && uv run check_rules.py

chat-claude ACTIVE 30000
chat-gpt ACTIVE 30000
chat-claude -> claude-haiku-5-5
chat-gpt    -> openai.gpt-5.6-luna
chat-claude ACTIVE 50
chat-gpt ACTIVE 50
chat-claude -> claude-haiku-4-5-20251001
chat-gpt    -> us.openai.gpt-5.6-luna

from strands import Agent

from gateway_models import claude_model, gpt_model
from tools import get_weather

# エージェントには論理名だけを渡す
for name, model in (("chat-claude", claude_model()), ("chat-gpt", gpt_model())):
    agent = Agent(model=model, tools=[get_weather], callback_handler=None)
    result = agent("東京と大阪の天気を教えて")
    calls = result.metrics.tool_metrics["get_weather"].call_count
    print(f"== {name}（ツール呼び出し {calls} 回）")
    print(str(result).strip(), "\n")

uv run setup_rules.py && uv run agent_logical.py

== chat-claude（ツール呼び出し 2 回）
東京と大阪の天気は、どちらも晴れで気温は18度です。 

== chat-gpt（ツール呼び出し 2 回）
- 東京：晴れ、18℃
- 大阪：晴れ、18℃

uv run setup_rules.py 50 && uv run agent_logical.py

== chat-claude（ツール呼び出し 2 回）
東京と大阪の天気情報です：
**東京**
- 天気：晴れ
- 気温：18度
**大阪**
- 天気：晴れ
- 気温：18度
両都市とも晴れており、気温も同じ18度です。過ごしやすい天気が続いているようです。

== chat-gpt（ツール呼び出し 2 回）
- 東京：晴れ、18℃
- 大阪：晴れ、18℃

import os
import time

import boto3

client = boto3.client("bedrock-agentcore-control", region_name="us-east-1")
gateway_id = os.environ["GATEWAY_URL"].removeprefix("https://").split(".")[0]

for rule in client.list_gateway_rules(gatewayIdentifier=gateway_id)["gatewayRules"]:
    client.delete_gateway_rule(gatewayIdentifier=gateway_id, ruleId=rule["ruleId"])
for target in client.list_gateway_targets(gatewayIdentifier=gateway_id)["items"]:
    client.delete_gateway_target(gatewayIdentifier=gateway_id, targetId=target["targetId"])
while client.list_gateway_targets(gatewayIdentifier=gateway_id)["items"]:
    time.sleep(5)
client.delete_gateway(gatewayIdentifier=gateway_id)
print("deleted", gateway_id)

uv run cleanup.py
aws iam delete-role-policy --role-name strands-gw-inference-role --policy-name inference
aws iam delete-role --role-name strands-gw-inference-role

from strands import Agent
from strands.models.anthropic import AnthropicModel

from jwt_token import GATEWAY_URL, get_token
from tools import get_weather

model = AnthropicModel(
    client_args={
        "base_url": f"{GATEWAY_URL}/inference",
        "auth_token": get_token(),  # Authorization: Bearer で送られる
    },
    model_id="us.anthropic.claude-haiku-5-5",
    max_tokens=1024,
)

agent = Agent(model=model, tools=[get_weather])
agent("東京と大阪の天気を教えて")

from strands import Agent
from strands.models.openai_responses import OpenAIResponsesModel

from jwt_token import GATEWAY_URL, get_token
from tools import get_weather

model = OpenAIResponsesModel(
    client_args={
        "base_url": f"{GATEWAY_URL}/inference/v1",
        "api_key": get_token(),  # Authorization: Bearer で送られる
    },
    model_id="openai.gpt-5.6-luna",
)

agent = Agent(model=model, tools=[get_weather])
agent("東京と大阪の天気を教えて")

uv run agent_claude_jwt.py
uv run agent_gpt_jwt.py