[Small Tip] Tried Calling AgentCore Gateway Inference Targets from Strands Agents

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

class GatewaySigV4Auth(httpx.Auth):
    """httpx Auth that attaches SigV4 signatures to requests to the Gateway"""

    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):
        # Remove headers for API keys added by the SDK since they're not used by the 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):
    """Continues to use the HTTP client without closing it even when the SDK's client is closed"""

    async def aclose(self) -> None:
        # Do nothing even if the SDK tries to close
        pass

    async def close_for_real(self) -> None:
        # Call this yourself when you're done to close
        await super().aclose()

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

from strands import tool

@tool
def get_weather(city: str) -> str:
    """Returns the weather for the specified city"""
    return f"{city} is sunny with a temperature of 18 degrees"

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("Tell me the weather in Tokyo and Osaka")

uv run agent_claude.py

Tool #1: get_weather

Tool #2: get_weather
The weather in both Tokyo and Osaka is sunny with a temperature of 18 degrees.

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("Tell me the weather in Tokyo and Osaka")
    finally:
        # Close the HTTP client when finished with the agent
        await http_client.close_for_real()

asyncio.run(main())

uv run agent_gpt.py

Tool #1: get_weather

Tool #2: get_weather
- Tokyo: Sunny, 18°C
- Osaka: Sunny, 18°C

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:
    """Model called in Anthropic Messages format"""
    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:
    """Model called in OpenAI Responses format"""
    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

# Determine primary and fallback destinations (models that can be called in the same format) for each logical name
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

# Pass only logical names to agents
for name, model in (("chat-claude", claude_model()), ("chat-gpt", gpt_model())):
    agent = Agent(model=model, tools=[get_weather], callback_handler=None)
    result = agent("Tell me the weather in Tokyo and Osaka")
    calls = result.metrics.tool_metrics["get_weather"].call_count
    print(f"== {name} (tool calls: {calls})")
    print(str(result).strip(), "\n")

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

== chat-claude (tool calls: 2)
The weather in both Tokyo and Osaka is sunny with a temperature of 18 degrees.

== chat-gpt (tool calls: 2)
- Tokyo: Sunny, 18°C
- Osaka: Sunny, 18°C

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

== chat-claude (tool calls: 2)
Here is the weather information for Tokyo and Osaka:
**Tokyo**
- Weather: Sunny
- Temperature: 18 degrees
**Osaka**
- Weather: Sunny
- Temperature: 18 degrees
Both cities are sunny with the same temperature of 18 degrees. It seems like comfortable weather continues.

== chat-gpt (tool calls: 2)
- Tokyo: Sunny, 18°C
- Osaka: Sunny, 18°C

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(),  # Sent as Authorization: Bearer
    },
    model_id="us.anthropic.claude-haiku-5-5",
    max_tokens=1024,
)

agent = Agent(model=model, tools=[get_weather])
agent("Tell me the weather in Tokyo and Osaka")

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(),  # Sent as Authorization: Bearer
    },
    model_id="openai.gpt-5.6-luna",
)

agent = Agent(model=model, tools=[get_weather])
agent("Tell me the weather in Tokyo and Osaka")

uv run agent_claude_jwt.py
uv run agent_gpt_jwt.py