6d42b876f2
DEFAULT_CONTEXT_LENGTHS declares claude-opus-5 as 1M, but the Bedrock static table never got the entry, so the offline path resolved the 128K default and the agent compressed context ~8x early. Add the key and pin the table pairing with a test so the next 1M generation cannot drift. Co-authored-by: JiaDe-Wu <JiaDe-Wu@users.noreply.github.com> (cherry picked from commit c1be16ecbdd267b14124dd35c0b92ccfcd201818)
1396 lines
70 KiB
Python
1396 lines
70 KiB
Python
"""AWS Bedrock Converse API adapter (boto3, optional dependency).
|
|
|
|
Works without API keys via the AWS credential chain, cross-region inference profiles, guardrails and
|
|
control-plane model discovery. OpenAI-format messages/tools are converted to Converse on the way in
|
|
and responses normalized back to OpenAI-shaped objects.
|
|
"""
|
|
|
|
from pm import install_hint
|
|
import base64
|
|
import importlib
|
|
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
import time
|
|
import traceback
|
|
from contextlib import suppress
|
|
from types import SimpleNamespace
|
|
from typing import Any, Dict, List, Optional, Tuple
|
|
from urllib.parse import urlparse
|
|
|
|
import httpx
|
|
|
|
from agent.errors import EmptyStreamError
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_bedrock_runtime_client_cache: Dict[str, Any] = {}
|
|
_bedrock_control_client_cache: Dict[str, Any] = {}
|
|
# Routed multiplex profiles: one client per (profile home, region). boto3 freezes the credential
|
|
# chain into the client at construction, so a region-only slot would sign profile B's calls with A's keys.
|
|
_bedrock_clients_by_home: Dict[Tuple[str, str, str], Any] = {}
|
|
|
|
# botocore session kwarg <- profile .env variable (the explicit sources of the default chain).
|
|
_AWS_SCOPED_CREDENTIAL_VARS: Tuple[Tuple[str, str], ...] = (
|
|
("aws_access_key_id", "AWS_ACCESS_KEY_ID"), ("aws_secret_access_key", "AWS_SECRET_ACCESS_KEY"),
|
|
("aws_session_token", "AWS_SESSION_TOKEN"), ("profile_name", "AWS_PROFILE"),
|
|
)
|
|
|
|
|
|
def scoped_aws_session_kwargs() -> Dict[str, str]:
|
|
"""``boto3.session.Session`` kwargs from the routed profile's secret scope, ``{}`` when unscoped.
|
|
|
|
Under a HERMES_HOME override the process env holds the LAUNCH profile's ``AWS_*`` (or nothing), so
|
|
every Bedrock client for a served profile must be built from that profile's own ``.env`` values.
|
|
Under multiplexing a profile that sets none of its own must NOT get ``{}`` — ``boto3.Session()``
|
|
would then resolve the ambient default chain (process env, ~/.aws, instance metadata), i.e. the
|
|
launch context's identity, exactly the borrow the Entra adapter refuses. ``AWS_PROFILE`` counts as
|
|
an explicit per-profile choice (it names an entry in the shared AWS config, like the Entra
|
|
``AZURE_CLIENT_ID``-only managed-identity opt-in)."""
|
|
from hermes_constants import get_hermes_home_override
|
|
if get_hermes_home_override() is None:
|
|
return {}
|
|
from agent.secret_scope import current_secret_scope, is_multiplex_active
|
|
scope = current_secret_scope() or {}
|
|
kwargs = {kw: scope[var].strip() for kw, var in _AWS_SCOPED_CREDENTIAL_VARS
|
|
if (scope.get(var) or "").strip()}
|
|
# Under multiplexing a partial set is not enough: boto3 fills whatever is missing from the
|
|
# ambient chain (process env, ~/.aws, instance metadata), i.e. the launch context's identity —
|
|
# the same borrow the Entra adapter refuses. Require a COMPLETE credential: the key pair, or
|
|
# AWS_PROFILE naming an entry in the shared AWS config (an explicit per-profile choice, like
|
|
# the Entra AZURE_CLIENT_ID-only managed-identity opt-in).
|
|
complete = ("aws_access_key_id" in kwargs and "aws_secret_access_key" in kwargs) \
|
|
or "profile_name" in kwargs
|
|
if not complete and is_multiplex_active():
|
|
raise RuntimeError(
|
|
"Bedrock auth is refused for this profile: it sets no complete AWS credential of its "
|
|
"own, and under multiplexed profiles the ambient default chain (process env, ~/.aws, "
|
|
"instance metadata) would sign this profile's calls with the LAUNCH context's identity. "
|
|
"Set AWS_ACCESS_KEY_ID + AWS_SECRET_ACCESS_KEY in this profile's own config, or "
|
|
"AWS_PROFILE to name a shared AWS config profile."
|
|
)
|
|
return kwargs
|
|
|
|
# Bedrock-hosted GPT-5.x models are served from the Bedrock Mantle OpenAI-compatible endpoint, not
|
|
# Converse. Narrow allowlist so GPT-OSS models stay on the native path.
|
|
BEDROCK_OPENAI_RESPONSES_MODEL_IDS: Tuple[str, ...] = (
|
|
"openai.gpt-5.5", "openai.gpt-5.6-sol", "openai.gpt-5.6-terra", "openai.gpt-5.6-luna",
|
|
)
|
|
_BEDROCK_OPENAI_HOST_RE = re.compile(r"^bedrock-mantle\.([a-z0-9-]+)\.api\.aws$", re.IGNORECASE)
|
|
# Bedrock-hosted xAI Grok (any regional inference-profile prefix) rejects temperature/topP in Converse
|
|
# with a hard 400 ("This model doesn't support the temperature field"); reasoning-first, same
|
|
# restriction as Claude Opus 4.6+ but _forbids_sampling_params is Claude-only, so it needs its own gate.
|
|
_BEDROCK_XAI_GROK_NO_SAMPLING_RE = re.compile(r"^(?:[a-z]+\.)?xai\.grok", re.IGNORECASE)
|
|
_MIN_BOTO3_VERSION = (1, 34, 59)
|
|
|
|
|
|
def _require_boto3():
|
|
"""Import boto3; converse_stream() needs >= 1.34.59 (a system boto3 can shadow the venv pin)."""
|
|
install_error = None
|
|
try:
|
|
# boto3 left [all] (PRs #24220, #24515); PM installs the [bedrock] extra on first use. This
|
|
# runs at the first client build, never at import: an import-time sync would rebuild the
|
|
# dependency environment of whatever process happens to import this module.
|
|
try:
|
|
from pm import ensure_import
|
|
ensure_import("bedrock")
|
|
except Exception as exc: # the import below decides; exc explains a miss
|
|
logger.warning("boto3 lazy install did not complete: %s", exc)
|
|
install_error = exc
|
|
import boto3
|
|
except ImportError:
|
|
# A completed install that needs a restart must not be reported as "install it".
|
|
reason = f": {install_error}" if install_error else f". Run: {install_hint('bedrock')}"
|
|
raise ImportError(f"The 'boto3' package is required for the AWS Bedrock provider{reason}") from install_error
|
|
try:
|
|
version = tuple(int(x) for x in boto3.__version__.split(".")[:3])
|
|
except (AttributeError, ValueError):
|
|
return boto3 # can't parse — don't block on version check
|
|
if version < _MIN_BOTO3_VERSION:
|
|
raise RuntimeError(
|
|
f"boto3 {boto3.__version__} does not support converse_stream "
|
|
f"(minimum 1.34.59 required). Run: hermes pm repair"
|
|
)
|
|
return boto3
|
|
|
|
|
|
def _cached_client(cache: Dict[str, Any], service: str, region: str):
|
|
"""Get or create a per-region boto3 client. Unscoped: the default credential chain, one client per
|
|
region. Routed profile: one client per (home, service, region), built from that profile's scoped
|
|
``AWS_*`` (falling back to the default chain only for what the profile does not set)."""
|
|
from hermes_constants import get_hermes_home_override, hermes_home_key
|
|
if get_hermes_home_override() is None:
|
|
if region not in cache:
|
|
cache[region] = _require_boto3().client(service, region_name=region)
|
|
return cache[region]
|
|
key = (hermes_home_key(), service, region)
|
|
client = _bedrock_clients_by_home.get(key)
|
|
if client is None:
|
|
# Scope check first: a cred-less multiplex profile must hit the ambient-chain
|
|
# refusal, not a boto3 ImportError on hosts that lack the package.
|
|
kwargs = scoped_aws_session_kwargs()
|
|
client = _require_boto3().Session(**kwargs).client(service, region_name=region)
|
|
_bedrock_clients_by_home[key] = client
|
|
return client
|
|
|
|
|
|
def _get_bedrock_runtime_client(region: str):
|
|
return _cached_client(_bedrock_runtime_client_cache, "bedrock-runtime", region)
|
|
|
|
|
|
def _get_bedrock_control_client(region: str):
|
|
return _cached_client(_bedrock_control_client_cache, "bedrock", region)
|
|
|
|
|
|
def reset_client_cache():
|
|
"""Clear cached boto3 clients. Used in tests and profile switches."""
|
|
_bedrock_runtime_client_cache.clear()
|
|
_bedrock_control_client_cache.clear()
|
|
_bedrock_clients_by_home.clear()
|
|
_inference_profile_model_cache.clear()
|
|
|
|
|
|
def invalidate_runtime_client(region: str) -> bool:
|
|
"""Evict one region's cached ``bedrock-runtime`` client (stale HTTP pool); True if evicted."""
|
|
from hermes_constants import get_hermes_home_override, hermes_home_key
|
|
if get_hermes_home_override() is not None:
|
|
return _bedrock_clients_by_home.pop((hermes_home_key(), "bedrock-runtime", region), None) is not None
|
|
return _bedrock_runtime_client_cache.pop(region, None) is not None
|
|
|
|
|
|
|
|
# --- Bedrock Mantle / OpenAI Responses support ---
|
|
|
|
def is_openai_bedrock_model(model_id: str) -> bool:
|
|
"""True for Bedrock-hosted OpenAI models that require Mantle (GPT-OSS excluded)."""
|
|
return str(model_id or "").strip().lower() in {m.lower() for m in BEDROCK_OPENAI_RESPONSES_MODEL_IDS}
|
|
|
|
|
|
def merge_bedrock_openai_model_ids(model_ids: List[str]) -> List[str]:
|
|
"""Append Mantle-only OpenAI models, which control-plane discovery never lists."""
|
|
merged = list(model_ids or [])
|
|
seen = {str(m).lower() for m in merged}
|
|
return merged + [m for m in BEDROCK_OPENAI_RESPONSES_MODEL_IDS if m.lower() not in seen]
|
|
|
|
|
|
def bedrock_openai_base_url(region: str) -> str:
|
|
"""Return Bedrock Mantle's OpenAI-compatible base URL for *region*."""
|
|
resolved = (region or "").strip() or resolve_bedrock_runtime_region()
|
|
return f"https://bedrock-mantle.{resolved}.api.aws/openai/v1"
|
|
|
|
|
|
def _mantle_url_parts(base_url: str) -> Tuple[Optional[str], str]:
|
|
"""(region or None if not a Mantle host, normalized path) for a base URL."""
|
|
parsed = urlparse(str(base_url or ""))
|
|
match = _BEDROCK_OPENAI_HOST_RE.match(parsed.hostname or "")
|
|
return (match.group(1) if match else None), (parsed.path or "").rstrip("/").lower()
|
|
|
|
|
|
def bedrock_openai_region_from_base_url(base_url: str) -> Optional[str]:
|
|
"""Extract the AWS region from a Bedrock Mantle OpenAI base URL."""
|
|
return _mantle_url_parts(base_url)[0]
|
|
|
|
|
|
def is_bedrock_openai_base_url(base_url: str) -> bool:
|
|
"""True for Bedrock Mantle endpoints (bare host or /openai[/v1] path)."""
|
|
region, path = _mantle_url_parts(base_url)
|
|
return region is not None and path in {"", "/openai", "/openai/v1"}
|
|
|
|
|
|
def resolve_bedrock_bearer_token(env: Optional[Dict[str, str]] = None) -> str:
|
|
"""Return AWS_BEARER_TOKEN_BEDROCK when Bedrock API-key auth is configured.
|
|
|
|
Under a HERMES_HOME override the read goes through the profile secret scope so a
|
|
served profile never inherits the launch profile's bearer from the process env."""
|
|
if env is not None:
|
|
return (env.get("AWS_BEARER_TOKEN_BEDROCK", "") or "").strip()
|
|
from hermes_constants import get_hermes_home_override
|
|
if get_hermes_home_override() is not None:
|
|
from agent.secret_scope import get_secret
|
|
return (get_secret("AWS_BEARER_TOKEN_BEDROCK", "") or "").strip()
|
|
return (os.environ.get("AWS_BEARER_TOKEN_BEDROCK", "") or "").strip()
|
|
|
|
|
|
class BedrockOpenAISigV4Auth(httpx.Auth):
|
|
"""httpx auth hook that SigV4-signs Bedrock Mantle OpenAI requests."""
|
|
|
|
requires_request_body = True
|
|
|
|
def __init__(self, region: str, service: str = "bedrock"):
|
|
self.region = (region or "").strip() or resolve_bedrock_runtime_region()
|
|
self.service = service
|
|
|
|
def auth_flow(self, request): # pragma: no cover - exercised by live call
|
|
from botocore.auth import SigV4Auth
|
|
from botocore.awsrequest import AWSRequest
|
|
kwargs = scoped_aws_session_kwargs()
|
|
credentials = _require_boto3().Session(**kwargs).get_credentials()
|
|
if credentials is None:
|
|
raise RuntimeError(
|
|
"No AWS credentials available for Bedrock OpenAI Responses. "
|
|
"Configure AWS_ACCESS_KEY_ID/AWS_SECRET_ACCESS_KEY, AWS_PROFILE, SSO, or an instance/task role."
|
|
)
|
|
# SigV4 must own Authorization: drop the SDK's placeholder bearer header.
|
|
headers = {str(k): str(v) for k, v in request.headers.items()
|
|
if str(k).lower() not in {"authorization", "x-amz-date", "x-amz-security-token"}}
|
|
aws_request = AWSRequest(method=request.method, url=str(request.url), data=request.content or b"", headers=headers)
|
|
SigV4Auth(credentials.get_frozen_credentials(), self.service, self.region).add_auth(aws_request)
|
|
request.headers.update(dict(aws_request.headers.items()))
|
|
yield request
|
|
|
|
|
|
def build_bedrock_openai_http_client(region: str, *, timeout: Optional[float] = None):
|
|
"""Build an httpx client that SigV4-signs Bedrock OpenAI requests."""
|
|
kwargs: Dict[str, Any] = {"auth": BedrockOpenAISigV4Auth(region)}
|
|
if isinstance(timeout, (int, float)) and not isinstance(timeout, bool) and timeout > 0:
|
|
kwargs["timeout"] = timeout
|
|
return httpx.Client(**kwargs)
|
|
|
|
|
|
def configure_bedrock_openai_client_kwargs(client_kwargs: Dict[str, Any], *, timeout: Optional[float] = None) -> Dict[str, Any]:
|
|
"""Install SigV4 auth on OpenAI SDK kwargs for Bedrock Mantle; a real API key keeps the SDK's
|
|
bearer auth, the ``aws-sdk``/``no-key-required`` placeholders mean IAM chain auth."""
|
|
base_url = str(client_kwargs.get("base_url") or "")
|
|
api_key = client_kwargs.get("api_key")
|
|
if not is_bedrock_openai_base_url(base_url) or (
|
|
isinstance(api_key, str) and api_key.strip() and api_key not in {"aws-sdk", "no-key-required"}
|
|
):
|
|
return client_kwargs
|
|
region = bedrock_openai_region_from_base_url(base_url) or resolve_bedrock_runtime_region()
|
|
client_kwargs["api_key"] = "aws-sdk"
|
|
client_kwargs["http_client"] = build_bedrock_openai_http_client(region, timeout=timeout)
|
|
return client_kwargs
|
|
|
|
|
|
# --- Stale-connection detection ---
|
|
# A pooled connection killed under boto3 (NAT timeout, VPN flap, RST) surfaces as a botocore/urllib3
|
|
# transport error or a bare AssertionError from urllib3's pool checks; retrying the same client
|
|
# reproduces it, so the fix is to evict the client.
|
|
|
|
_STALE_LIB_MODULE_PREFIXES = ("urllib3.", "botocore.", "boto3.")
|
|
|
|
|
|
def _stale_error_types() -> tuple:
|
|
"""botocore + urllib3 transport-failure exception classes (best-effort import)."""
|
|
types: list = []
|
|
for module, names in (
|
|
("botocore.exceptions", ("ConnectionError", "HTTPClientError")),
|
|
("urllib3.exceptions", ("ProtocolError", "NewConnectionError", "ConnectionError")),
|
|
):
|
|
# AttributeError too: ``from mod import Name`` raised ImportError for a missing name.
|
|
with suppress(ImportError, AttributeError): # pragma: no cover — both present with boto3
|
|
types += [getattr(importlib.import_module(module), name) for name in names]
|
|
return tuple(types)
|
|
|
|
|
|
def is_stale_connection_error(exc: BaseException) -> bool:
|
|
"""True for botocore/urllib3 transport errors or AssertionErrors raised inside those libs."""
|
|
if isinstance(exc, _stale_error_types()):
|
|
return True
|
|
return isinstance(exc, AssertionError) and any(
|
|
(frame.f_globals.get("__name__", "") or "").startswith(_STALE_LIB_MODULE_PREFIXES)
|
|
for frame, _lineno in traceback.walk_tb(exc.__traceback__)
|
|
)
|
|
|
|
|
|
def is_streaming_access_denied_error(exc: BaseException) -> bool:
|
|
"""True when IAM denied ``bedrock:InvokeModelWithResponseStream`` (permanent: callers fall back
|
|
to converse()). Message-based: the AnthropicBedrock SDK wraps the response but keeps the action name."""
|
|
msg = str(exc).lower()
|
|
if "invokemodelwithresponsestream" not in msg:
|
|
return False
|
|
with suppress(ImportError): # pragma: no cover — botocore always present with boto3
|
|
from botocore.exceptions import ClientError
|
|
if isinstance(exc, ClientError):
|
|
code = (getattr(exc, "response", None) or {}).get("Error", {}).get("Code", "")
|
|
return code in ("AccessDeniedException", "UnauthorizedException")
|
|
return "not authorized" in msg or "accessdenied" in msg
|
|
|
|
|
|
# --- AWS credential detection ---
|
|
# Priority order; the first group whose vars are ALL set names the auth source.
|
|
_AWS_AUTH_ENV_CHAIN: Tuple[Tuple[str, ...], ...] = (
|
|
("AWS_BEARER_TOKEN_BEDROCK",), # Bedrock bearer token
|
|
("AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY"), # explicit IAM key pair
|
|
("AWS_PROFILE",), # named profile (SSO, assume-role)
|
|
("AWS_CONTAINER_CREDENTIALS_RELATIVE_URI",), # ECS / CodeBuild
|
|
("AWS_WEB_IDENTITY_TOKEN_FILE",), # EKS IRSA
|
|
)
|
|
|
|
|
|
def _boto3_chain_has_credentials() -> bool:
|
|
"""True if boto3's default chain resolves credentials (IMDS, task role, ...)."""
|
|
with suppress(Exception):
|
|
import botocore.session
|
|
credentials = botocore.session.get_session().get_credentials()
|
|
resolved = credentials.get_frozen_credentials() if credentials is not None else None
|
|
return bool(resolved and resolved.access_key)
|
|
return False
|
|
|
|
|
|
def resolve_aws_auth_env_var(env: Optional[Dict[str, str]] = None) -> Optional[str]:
|
|
"""Name of the active AWS auth source: env vars first (no I/O), then ``"iam-role"`` via boto3's chain, else None."""
|
|
env = env if env is not None else os.environ
|
|
for group in _AWS_AUTH_ENV_CHAIN:
|
|
if all(env.get(var, "").strip() for var in group):
|
|
return group[0]
|
|
return "iam-role" if _boto3_chain_has_credentials() else None
|
|
|
|
|
|
def has_aws_credentials(env: Optional[Dict[str, str]] = None) -> bool:
|
|
"""True if any AWS credential source (env vars or boto3 chain) is detected.
|
|
|
|
This two-tier approach mirrors the pattern from OpenClaw PR #62673: cloud environments (EC2, ECS,
|
|
Lambda) provide credentials via instance metadata, not environment variables. The env-var check is a
|
|
fast path for local development; the boto3 fallback covers all cloud deployments.
|
|
"""
|
|
return resolve_aws_auth_env_var(env) is not None or _boto3_chain_has_credentials()
|
|
|
|
|
|
def resolve_bedrock_region(env: Optional[Dict[str, str]] = None) -> str:
|
|
"""AWS_REGION → AWS_DEFAULT_REGION → botocore configured region (~/.aws/config profiles) → us-east-1."""
|
|
env = env if env is not None else os.environ
|
|
explicit = env.get("AWS_REGION", "").strip() or env.get("AWS_DEFAULT_REGION", "").strip()
|
|
if explicit:
|
|
return explicit
|
|
with suppress(Exception):
|
|
import botocore.session
|
|
return botocore.session.get_session().get_config_variable("region") or "us-east-1"
|
|
return "us-east-1"
|
|
|
|
|
|
def resolve_bedrock_runtime_region(config: Optional[Dict[str, Any]] = None) -> str:
|
|
"""``bedrock.region`` from config.yaml, else :func:`resolve_bedrock_region`. Every non-runtime Bedrock
|
|
endpoint must use this so auxiliary calls never leave the primary runtime's region. *config* skips disk."""
|
|
if config is None:
|
|
with suppress(Exception):
|
|
from hermes_cli.config import load_config_readonly
|
|
config = load_config_readonly()
|
|
cfg_region = str(((config or {}).get("bedrock") or {}).get("region") or "").strip()
|
|
return cfg_region or resolve_bedrock_region()
|
|
|
|
|
|
def bedrock_region_from_runtime_url(base_url: str) -> str:
|
|
"""AWS region from a ``bedrock-runtime.<region>.amazonaws.com`` URL (default us-east-1)."""
|
|
m = re.search(r"bedrock-runtime\.([a-z0-9-]+)\.", base_url or "")
|
|
return m.group(1) if m else "us-east-1"
|
|
|
|
|
|
def bedrock_guardrail_config(config: Optional[Dict[str, Any]] = None) -> Optional[Dict[str, Any]]:
|
|
"""Converse ``guardrailConfig`` from ``bedrock.guardrail`` in config.yaml (None when unset)."""
|
|
if config is None:
|
|
config = {}
|
|
with suppress(Exception):
|
|
from hermes_cli.config import load_config_readonly
|
|
config = load_config_readonly()
|
|
gr = ((config or {}).get("bedrock") or {}).get("guardrail") or {}
|
|
if not (gr.get("guardrail_identifier") and gr.get("guardrail_version")):
|
|
return None
|
|
out = {"guardrailIdentifier": gr["guardrail_identifier"], "guardrailVersion": gr["guardrail_version"]}
|
|
for src, dst in (("stream_processing_mode", "streamProcessingMode"), ("trace", "trace")):
|
|
if gr.get(src):
|
|
out[dst] = gr[src]
|
|
return out
|
|
|
|
|
|
def bedrock_guardrail_headers(config: Optional[Dict[str, Any]] = None) -> Dict[str, str]:
|
|
"""InvokeModel/Messages-wire form of the configured guardrail. The AnthropicBedrock SDK speaks
|
|
InvokeModel, which has no ``guardrailConfig`` body field; Bedrock reads the guardrail from these
|
|
headers instead (same enforcement, keeps prompt caching / thinking / 1M context)."""
|
|
gr = bedrock_guardrail_config(config)
|
|
if not gr:
|
|
return {}
|
|
headers = {
|
|
"X-Amzn-Bedrock-GuardrailIdentifier": str(gr["guardrailIdentifier"]),
|
|
"X-Amzn-Bedrock-GuardrailVersion": str(gr["guardrailVersion"]),
|
|
}
|
|
if str(gr.get("trace", "")).lower() in {"enabled", "enabled_full", "true"}:
|
|
headers["X-Amzn-Bedrock-Trace"] = "ENABLED"
|
|
return headers
|
|
|
|
|
|
GUARDRAIL_ACTION_FIELD = "amazon-bedrock-guardrailAction"
|
|
|
|
|
|
def anthropic_response_guardrail_intervened(response: Any) -> bool:
|
|
"""True when Bedrock substituted the InvokeModel reply with guardrail messaging. Unlike Converse
|
|
(``stopReason=guardrail_intervened``), InvokeModel keeps ``stop_reason=end_turn`` and signals the
|
|
block only via an unmodelled body field the Anthropic SDK keeps in ``model_extra``."""
|
|
extra = getattr(response, "model_extra", None) or {}
|
|
return str(extra.get(GUARDRAIL_ACTION_FIELD, "")).upper() == "INTERVENED"
|
|
|
|
|
|
def bind_bedrock_runtime(agent, base_url: str, api_mode: str) -> None:
|
|
"""Point *agent* at a non-Mantle Bedrock wire: ``bedrock_converse`` (boto3 direct, no SDK client) or
|
|
``anthropic_messages`` (AnthropicBedrock SDK, SigV4 via the boto3 chain). ``aws-sdk`` is a sentinel,
|
|
never a credential, so the generic Anthropic/OpenAI client builders must not see it. Startup and every
|
|
later rebuild (/model switch, fallback restore, fallback-to-Bedrock) share this so region and guardrail
|
|
state never lag the active endpoint."""
|
|
agent._bedrock_region = bedrock_region_from_runtime_url(base_url)
|
|
agent._bedrock_guardrail_config = bedrock_guardrail_config()
|
|
agent.client = None
|
|
agent._client_kwargs = {}
|
|
agent.api_key = agent._anthropic_api_key = "aws-sdk"
|
|
agent._anthropic_base_url = base_url
|
|
agent._is_anthropic_oauth = False
|
|
if api_mode == "anthropic_messages":
|
|
from agent.anthropic_adapter import build_anthropic_bedrock_client
|
|
agent._anthropic_client = build_anthropic_bedrock_client(agent._bedrock_region)
|
|
else:
|
|
agent._anthropic_client = None
|
|
|
|
|
|
def bedrock_model_ids_or_none() -> Optional[List[str]]:
|
|
"""Live-discover Bedrock model IDs; None on failure/empty so callers use the static list."""
|
|
with suppress(Exception):
|
|
discovered = discover_bedrock_models(resolve_bedrock_runtime_region())
|
|
if discovered:
|
|
return merge_bedrock_openai_model_ids([m["id"] for m in discovered])
|
|
return None
|
|
|
|
|
|
# --- Tool-calling / prompt-cache capability detection ---
|
|
# Models known to reject toolConfig with a ValidationException; unknown models assumed OK.
|
|
_NON_TOOL_CALLING_PATTERNS = [
|
|
"deepseek.r1", "deepseek-r1", # DeepSeek R1 (both ID formats) — reasoning only
|
|
"stability.", # image generation
|
|
"cohere.embed", "amazon.titan-embed", # embeddings
|
|
]
|
|
|
|
# cachePoint allowlist — inverted policy vs tools: unknown models get NO cache markers (they reject
|
|
# cachePoint). Claude only reaches build_converse_kwargs under bearer auth.
|
|
_CACHE_POINT_PATTERNS = ["anthropic.claude", "amazon.nova"]
|
|
|
|
|
|
def _model_supports_tool_use(model_id: str) -> bool:
|
|
"""False for denylisted models; unknown models default to True."""
|
|
return not any(pattern in model_id.lower() for pattern in _NON_TOOL_CALLING_PATTERNS)
|
|
|
|
|
|
def _model_supports_prompt_cache(model_id: str) -> bool:
|
|
# An application-inference-profile ARN names no model: match on the wrapped model (cached lookup).
|
|
if _APPLICATION_PROFILE_ARN_RE.search(model_id):
|
|
model_id = _resolve_inference_profile_model_id(model_id)
|
|
return any(pattern in model_id.lower() for pattern in _CACHE_POINT_PATTERNS)
|
|
|
|
|
|
# --- Server-verdict cachePoint suppression ---
|
|
# Bedrock's cachePoint rule is per-family AND per-field (Nova accepts it in system/messages but hard-fails
|
|
# on toolConfig.tools) and any static table drifts, so when Bedrock names a placement as unpermitted we
|
|
# record the verdict, drop the marker there for the rest of the process, and retry once without it.
|
|
|
|
CACHE_POINT_PLACEMENTS = ("tools", "system", "messages")
|
|
_CACHE_POINT_REJECTIONS: Dict[str, set] = {} # model_id (lowercased) → placements Bedrock has rejected this process
|
|
# e.g. "#/toolConfig/tools/18: extraneous key [cachePoint] is not permitted"
|
|
_CACHE_POINT_PATH_PATTERN = re.compile(r"#/(?P<path>[A-Za-z0-9_./\[\]-]*)", re.IGNORECASE)
|
|
_CACHE_POINT = {"cachePoint": {"type": "default"}}
|
|
|
|
|
|
def cache_point_rejection_placement(exc: BaseException) -> Optional[str]:
|
|
"""Converse section whose cachePoint Bedrock refused, or None. Message-based: the JSON pointer in the
|
|
ValidationException is the only thing naming the section (raw or SDK-wrapped). Unlocalisable → "tools"."""
|
|
msg = str(exc)
|
|
lowered = msg.lower()
|
|
if "cachepoint" not in lowered or ("not permitted" not in lowered and "extraneous" not in lowered):
|
|
return None
|
|
match = _CACHE_POINT_PATH_PATTERN.search(msg)
|
|
path = (match.group("path") if match else "").lower()
|
|
if "toolconfig" in path or "tools" in path:
|
|
return "tools"
|
|
return next((placement for placement in ("system", "messages") if placement in path), "tools")
|
|
|
|
|
|
def note_cache_point_rejection(model_id: str, placement: str) -> None:
|
|
"""Record that ``model_id`` refuses cachePoint blocks in ``placement``."""
|
|
if placement in CACHE_POINT_PLACEMENTS:
|
|
_CACHE_POINT_REJECTIONS.setdefault(model_id.lower(), set()).add(placement)
|
|
|
|
|
|
def cache_point_allowed(model_id: str, placement: str) -> bool:
|
|
"""False once Bedrock has refused this placement for this model."""
|
|
return placement not in _CACHE_POINT_REJECTIONS.get(model_id.lower(), ())
|
|
|
|
|
|
def reset_cache_point_rejections() -> None:
|
|
"""Clear recorded cachePoint rejections. Used in tests."""
|
|
_CACHE_POINT_REJECTIONS.clear()
|
|
|
|
|
|
def _without_cache_points(blocks: Any) -> Optional[list]:
|
|
"""``blocks`` minus cachePoint entries, or None if not a list / nothing removed."""
|
|
if not isinstance(blocks, list):
|
|
return None
|
|
cleaned = [b for b in blocks if not (isinstance(b, dict) and set(b.keys()) == {"cachePoint"})]
|
|
return None if len(cleaned) == len(blocks) else cleaned
|
|
|
|
|
|
def strip_cache_points(kwargs: Dict[str, Any], placement: str) -> Dict[str, Any]:
|
|
"""Copy of Converse kwargs with ``placement``'s cachePoint removed; the SAME object
|
|
back when nothing was stripped (callers use identity to decide a retry cannot help)."""
|
|
if placement == "messages":
|
|
messages = kwargs.get("messages")
|
|
cleaned_contents = [
|
|
_without_cache_points(msg.get("content") if isinstance(msg, dict) else None) for msg in messages
|
|
] if isinstance(messages, list) else []
|
|
if all(content is None for content in cleaned_contents):
|
|
return kwargs
|
|
return {**kwargs, "messages": [
|
|
msg if content is None else {**msg, "content": content} for msg, content in zip(messages, cleaned_contents)
|
|
]}
|
|
if placement == "system":
|
|
cleaned = _without_cache_points(kwargs.get("system"))
|
|
return kwargs if cleaned is None else {**kwargs, "system": cleaned}
|
|
if placement == "tools":
|
|
tool_config = kwargs.get("toolConfig")
|
|
cleaned = _without_cache_points((tool_config or {}).get("tools"))
|
|
return kwargs if cleaned is None else {**kwargs, "toolConfig": {**tool_config, "tools": cleaned}}
|
|
return kwargs
|
|
|
|
|
|
def recover_from_cache_point_rejection(exc: BaseException, kwargs: Dict[str, Any]) -> Optional[Dict[str, Any]]:
|
|
"""Record Bedrock's cachePoint verdict and return retry kwargs, or None when the error
|
|
was not a cachePoint rejection / the marker was already absent (caller re-raises)."""
|
|
placement = cache_point_rejection_placement(exc)
|
|
if placement is None:
|
|
return None
|
|
retry_kwargs = strip_cache_points(kwargs, placement)
|
|
if retry_kwargs is kwargs:
|
|
return None
|
|
model_id = str(kwargs.get("modelId", ""))
|
|
note_cache_point_rejection(model_id, placement)
|
|
logger.warning(
|
|
"bedrock: %s rejected a cachePoint block in %s — dropping that cache marker for this model and "
|
|
"retrying. Prompt caching stays active for the remaining sections.", model_id or "model", placement,
|
|
)
|
|
return retry_kwargs
|
|
|
|
|
|
# --- Encrypted-content redacted-reasoning suppression ---
|
|
# Redacted thinking blobs are sealed to the issuing model/flow; replaying them after a model switch or
|
|
# across regions fails with an encrypted-content ValidationException. Drop the redacted blocks and
|
|
# resend once (mirrors the cachePoint self-heal above: same-object return means no retry can help).
|
|
_REDACTED_REASONING_REJECTION_PATTERN = re.compile(
|
|
r"ValidationException.*(?:redacted|encrypt)", re.IGNORECASE | re.DOTALL,
|
|
)
|
|
|
|
|
|
def _without_redacted_reasoning(blocks):
|
|
"""``blocks`` minus reasoningContent entries carrying redactedContent, or None if not a list /
|
|
nothing removed. Pure reasoningText blocks are kept."""
|
|
if not isinstance(blocks, list):
|
|
return None
|
|
cleaned = [b for b in blocks if not (
|
|
isinstance(b, dict) and isinstance(b.get("reasoningContent"), dict)
|
|
and "redactedContent" in b["reasoningContent"]
|
|
)]
|
|
return None if len(cleaned) == len(blocks) else cleaned
|
|
|
|
|
|
def strip_redacted_reasoning(kwargs):
|
|
"""Copy of Converse kwargs with redacted reasoning blocks removed; the SAME object
|
|
back when nothing was stripped (callers use identity to decide a retry cannot help).
|
|
Turns left with empty content are dropped (they carried no replayable signal)."""
|
|
messages = kwargs.get("messages")
|
|
if not isinstance(messages, list):
|
|
return kwargs
|
|
cleaned_contents = [
|
|
_without_redacted_reasoning(msg.get("content") if isinstance(msg, dict) else None)
|
|
for msg in messages
|
|
]
|
|
if all(content is None for content in cleaned_contents):
|
|
return kwargs
|
|
return {**kwargs, "messages": [
|
|
msg if content is None else {**msg, "content": content}
|
|
for msg, content in zip(messages, cleaned_contents)
|
|
if content is None or len(content) > 0
|
|
]}
|
|
|
|
|
|
def recover_from_redacted_reasoning_rejection(exc, kwargs):
|
|
"""Return retry kwargs with redacted reasoning blocks stripped, or None when the error
|
|
was not an encrypted-content rejection / nothing redacted remained (caller re-raises)."""
|
|
if not _REDACTED_REASONING_REJECTION_PATTERN.search(str(exc)):
|
|
return None
|
|
retry_kwargs = strip_redacted_reasoning(kwargs)
|
|
if retry_kwargs is kwargs:
|
|
return None
|
|
logger.warning(
|
|
"bedrock: %s rejected replayed redacted reasoning (encrypted content is sealed to the issuing "
|
|
"model/flow) - stripping redacted blocks and resending once.",
|
|
str(kwargs.get("modelId", "")) or "model",
|
|
)
|
|
return retry_kwargs
|
|
|
|
|
|
# One optional regional/global inference-profile prefix, then the Claude model family.
|
|
_ANTHROPIC_BEDROCK_MODEL_RE = re.compile(
|
|
r"^(?:(?:global|us|eu|apac|ap|au|jp|ca|sa|me|af)\.)?anthropic\.claude", re.IGNORECASE,
|
|
)
|
|
|
|
|
|
def is_anthropic_bedrock_model(model_id: str) -> bool:
|
|
"""True for Claude on Bedrock (``anthropic.claude-*``, any regional prefix): AnthropicBedrock SDK path."""
|
|
return _ANTHROPIC_BEDROCK_MODEL_RE.match(model_id) is not None
|
|
|
|
|
|
# --- Message format conversion: OpenAI → Bedrock Converse ---
|
|
|
|
def convert_tools_to_converse(tools: List[Dict]) -> List[Dict]:
|
|
"""OpenAI ``{"function": {...}}`` tool defs → Converse ``{"toolSpec": {...}}``."""
|
|
return [{"toolSpec": {
|
|
"name": fn.get("name", ""), "description": fn.get("description", ""),
|
|
"inputSchema": {"json": fn.get("parameters", {"type": "object", "properties": {}})},
|
|
}} for fn in (t.get("function", {}) for t in tools or [])]
|
|
|
|
|
|
# Converse rejects empty OR whitespace-only text blocks, so the placeholder must be non-whitespace.
|
|
# A lone space is whitespace and is rejected too — the placeholder MUST itself be non-whitespace. Ref: issue
|
|
# #9486.
|
|
_EMPTY_TEXT_PLACEHOLDER = "(empty)"
|
|
_PLACEHOLDER_BLOCK = {"text": _EMPTY_TEXT_PLACEHOLDER}
|
|
|
|
|
|
def _safe_text(text) -> str:
|
|
"""``text`` if it has non-whitespace content, else the placeholder (None/non-str ok)."""
|
|
text = "" if text is None else str(text)
|
|
return text if text.strip() else _EMPTY_TEXT_PLACEHOLDER
|
|
|
|
|
|
def _image_block_from_data_url(url: str) -> Dict:
|
|
"""``data:<mime>;base64,...`` → Converse image block with RAW bytes (boto3 base64-encodes on the
|
|
wire; passing the string through double-encodes and Bedrock rejects it)."""
|
|
header, _, data = url.partition(",")
|
|
media_type = (header[5:].split(";")[0] if header.startswith("data:") else "") or "image/jpeg"
|
|
try:
|
|
# Ref: #33317.
|
|
raw_bytes = base64.b64decode(data)
|
|
except Exception:
|
|
raw_bytes = data.encode("utf-8")
|
|
return {"image": {"format": media_type.split("/")[-1] if "/" in media_type else "jpeg", "source": {"bytes": raw_bytes}}}
|
|
|
|
|
|
def _convert_content_to_converse(content) -> List[Dict]:
|
|
"""OpenAI content → Converse blocks; blank text → placeholder, remote image URLs → text reference."""
|
|
if not isinstance(content, list):
|
|
return [{"text": _safe_text(content)}]
|
|
blocks = []
|
|
for part in content:
|
|
if isinstance(part, str):
|
|
blocks.append({"text": _safe_text(part)})
|
|
elif isinstance(part, dict) and part.get("type", "") == "text":
|
|
blocks.append({"text": _safe_text(part.get("text", ""))})
|
|
elif isinstance(part, dict) and part.get("type", "") == "image_url":
|
|
image_url = part.get("image_url", {})
|
|
url = image_url.get("url", "") if isinstance(image_url, dict) else ""
|
|
blocks.append(_image_block_from_data_url(url) if url.startswith("data:") else {"text": f"[Image: {url}]"})
|
|
return blocks or [dict(_PLACEHOLDER_BLOCK)]
|
|
|
|
|
|
def _system_blocks(content) -> List[Dict]:
|
|
"""System content → text blocks; blank parts are dropped, not placeholder-filled."""
|
|
parts = [content] if isinstance(content, str) else content if isinstance(content, list) else []
|
|
texts = [part.get("text", "") if isinstance(part, dict) and part.get("type") == "text" else part for part in parts]
|
|
return [{"text": text} for text in texts if isinstance(text, str) and text.strip()]
|
|
|
|
|
|
def _tool_use_block(tool_use_id, name, input_dict) -> Dict:
|
|
return {"toolUse": {"toolUseId": tool_use_id, "name": name, "input": input_dict}}
|
|
|
|
|
|
def _decode_redacted(encoded) -> Optional[bytes]:
|
|
"""Strict base64 → bytes; None for empty/non-str/undecodable input."""
|
|
try:
|
|
return base64.b64decode(encoded, validate=True) if isinstance(encoded, str) and encoded else None
|
|
except (ValueError, TypeError):
|
|
return None
|
|
|
|
|
|
def _replay_ordered_blocks(ordered_blocks: List) -> List[Dict]:
|
|
"""Rebuild the exact Bedrock block sequence captured at normalization time; redacted reasoning is
|
|
stored base64 (JSON-safe sidecar) and undecodable entries are skipped."""
|
|
content_blocks: List[Dict] = []
|
|
for block in ordered_blocks:
|
|
if not isinstance(block, dict):
|
|
continue
|
|
if "text" in block and isinstance(block["text"], str):
|
|
content_blocks.append({"text": block["text"]})
|
|
elif "reasoningContent" in block:
|
|
reasoning = block["reasoningContent"]
|
|
if not isinstance(reasoning, dict):
|
|
continue
|
|
# ReasoningContentBlock is a tagged union: reasoningText and redactedContent must go
|
|
# out as separate blocks (#115865). Undecodable redacted entries are skipped alone.
|
|
if isinstance(reasoning.get("text"), str):
|
|
reasoning_text: Dict[str, str] = {"text": reasoning["text"]}
|
|
if isinstance(reasoning.get("signature"), str) and reasoning["signature"]:
|
|
reasoning_text["signature"] = reasoning["signature"] # models that sign thinking reject unsigned replay
|
|
content_blocks.append({"reasoningContent": {"reasoningText": reasoning_text}})
|
|
encoded = reasoning.get("redactedContentBase64")
|
|
if isinstance(encoded, str) and encoded:
|
|
redacted = _decode_redacted(encoded)
|
|
if redacted is not None:
|
|
content_blocks.append({"reasoningContent": {"redactedContent": redacted}})
|
|
elif "toolUse" in block and isinstance(block["toolUse"], dict):
|
|
tu = block["toolUse"]
|
|
content_blocks.append(_tool_use_block(tu.get("toolUseId", ""), tu.get("name", ""), tu.get("input", {})))
|
|
return content_blocks
|
|
|
|
|
|
def _parse_tool_args(args) -> Any:
|
|
"""JSON-decode a tool-call argument string; {} on failure; non-str passes through."""
|
|
try:
|
|
return json.loads(args) if isinstance(args, str) else args
|
|
except (json.JSONDecodeError, TypeError):
|
|
return {}
|
|
|
|
|
|
def _assistant_blocks(msg: Dict, content) -> List[Dict]:
|
|
"""Assistant message → Converse blocks. An ordered ``bedrock_content_blocks`` sidecar is authoritative;
|
|
otherwise redacted thinking from ``reasoning_details`` (byte-for-byte), then text, then tool calls."""
|
|
ordered_blocks = msg.get("bedrock_content_blocks")
|
|
if isinstance(ordered_blocks, list) and (content_blocks := _replay_ordered_blocks(ordered_blocks)):
|
|
return content_blocks
|
|
redacted = [
|
|
_decode_redacted(d.get("data") or d.get("redactedContentBase64"))
|
|
for d in (msg.get("reasoning_details") or []) if isinstance(d, dict) and d.get("type") == "redacted_thinking"
|
|
]
|
|
content_blocks: List[Dict] = [{"reasoningContent": {"redactedContent": r}} for r in redacted if r is not None]
|
|
if isinstance(content, str) and content.strip():
|
|
content_blocks.append({"text": content})
|
|
elif isinstance(content, list):
|
|
content_blocks.extend(_convert_content_to_converse(content))
|
|
for tc in (msg.get("tool_calls", []) or []):
|
|
fn = tc.get("function", {})
|
|
content_blocks.append(_tool_use_block(tc.get("id", ""), fn.get("name", ""), _parse_tool_args(fn.get("arguments", "{}"))))
|
|
return content_blocks
|
|
|
|
|
|
def convert_messages_to_converse(messages: List[Dict]) -> Tuple[Optional[List[Dict]], List[Dict]]:
|
|
"""OpenAI messages → ``(system_blocks_or_None, converse_messages)``; tool results become ``toolResult``
|
|
user blocks. Converse needs strict user/assistant alternation with a user turn first and last:
|
|
same-role neighbours merge, placeholder user turns pad the ends."""
|
|
system_blocks: List[Dict] = []
|
|
converse_msgs: List[Dict] = []
|
|
|
|
def append_turn(role: str, blocks: List[Dict]) -> None:
|
|
if converse_msgs and converse_msgs[-1]["role"] == role:
|
|
converse_msgs[-1]["content"].extend(blocks)
|
|
else:
|
|
converse_msgs.append({"role": role, "content": blocks})
|
|
|
|
for msg in messages:
|
|
role = msg.get("role", "")
|
|
content = msg.get("content")
|
|
if role == "system":
|
|
system_blocks.extend(_system_blocks(content))
|
|
elif role == "tool":
|
|
result_content = content if isinstance(content, str) else json.dumps(content)
|
|
append_turn("user", [{"toolResult": {
|
|
"toolUseId": msg.get("tool_call_id", ""), "content": [{"text": _safe_text(result_content)}]}}])
|
|
elif role == "assistant":
|
|
append_turn("assistant", _assistant_blocks(msg, content) or [dict(_PLACEHOLDER_BLOCK)])
|
|
elif role == "user":
|
|
append_turn("user", _convert_content_to_converse(content))
|
|
if converse_msgs and converse_msgs[0]["role"] != "user":
|
|
converse_msgs.insert(0, {"role": "user", "content": [dict(_PLACEHOLDER_BLOCK)]})
|
|
if converse_msgs and converse_msgs[-1]["role"] != "user":
|
|
converse_msgs.append({"role": "user", "content": [dict(_PLACEHOLDER_BLOCK)]})
|
|
return (system_blocks or None, converse_msgs)
|
|
|
|
|
|
# --- Response format conversion: Bedrock Converse → OpenAI ---
|
|
|
|
# Bedrock stopReason → OpenAI finish_reason (unknown → "stop").
|
|
_STOP_REASON_TO_FINISH_REASON = {
|
|
"end_turn": "stop", "stop_sequence": "stop", "tool_use": "tool_calls", "max_tokens": "length",
|
|
"content_filtered": "content_filter", "guardrail_intervened": "content_filter",
|
|
}
|
|
|
|
|
|
def _encode_redacted(redacted) -> Optional[str]:
|
|
"""Redacted reasoning payload → base64 str (bytes encoded, str passed through, else None)."""
|
|
if isinstance(redacted, (bytes, bytearray)):
|
|
return base64.b64encode(bytes(redacted)).decode("ascii")
|
|
return redacted if isinstance(redacted, str) else None
|
|
|
|
|
|
def _tool_call_ns(tool_use_id: str, name: str, input_dict) -> SimpleNamespace:
|
|
return SimpleNamespace(
|
|
id=tool_use_id, type="function", function=SimpleNamespace(name=name, arguments=json.dumps(input_dict)),
|
|
)
|
|
|
|
|
|
class _ResponseParts:
|
|
"""Accumulator shared by the sync and streaming normalizers."""
|
|
|
|
def __init__(self) -> None:
|
|
self.text_parts: List[str] = []
|
|
self.reasoning_parts: List[str] = []
|
|
self.reasoning_details: List[Dict[str, Any]] = []
|
|
self.tool_calls: List[SimpleNamespace] = []
|
|
|
|
def absorb_reasoning(self, reasoning: Any, block: Dict[str, Any], on_text=None) -> None:
|
|
"""Fold a Converse ``reasoningContent`` payload into the accumulators and ``block``. The sync response
|
|
nests ``reasoningText: {text, signature}``; stream deltas carry ``text`` / ``signature`` flat."""
|
|
if not isinstance(reasoning, dict):
|
|
return
|
|
if isinstance(reasoning.get("reasoningText"), dict):
|
|
reasoning = {**reasoning, **reasoning["reasoningText"]}
|
|
thinking_text = reasoning.get("text", "")
|
|
if thinking_text:
|
|
self.reasoning_parts.append(str(thinking_text))
|
|
if on_text:
|
|
on_text(thinking_text)
|
|
block["text"] = block.get("text", "") + str(thinking_text)
|
|
signature = reasoning.get("signature")
|
|
if isinstance(signature, str) and signature:
|
|
block["signature"] = block.get("signature", "") + signature
|
|
encoded = _encode_redacted(reasoning.get("redactedContent"))
|
|
if encoded:
|
|
self.reasoning_details.append({"type": "redacted_thinking", "data": encoded})
|
|
block["redactedContentBase64"] = encoded
|
|
|
|
def build(self, ordered_blocks: List[Dict[str, Any]], usage_data: Dict[str, int], stop_reason: str, model: str) -> SimpleNamespace:
|
|
"""Assemble the OpenAI-shaped response. Converse's inputTokens EXCLUDES cache read/write tokens
|
|
(OpenAI's prompt_tokens includes them), so they are added back."""
|
|
msg = SimpleNamespace(
|
|
role="assistant", content="\n".join(self.text_parts) if self.text_parts else None,
|
|
tool_calls=self.tool_calls or None, reasoning_details=self.reasoning_details or None,
|
|
reasoning_content="\n\n".join(self.reasoning_parts) if self.reasoning_parts else None,
|
|
bedrock_content_blocks=ordered_blocks or None,
|
|
)
|
|
cache_read_tokens, cache_write_tokens, output_tokens = (
|
|
usage_data.get(k, 0) for k in ("cacheReadInputTokens", "cacheWriteInputTokens", "outputTokens")
|
|
)
|
|
prompt_tokens = usage_data.get("inputTokens", 0) + cache_read_tokens + cache_write_tokens
|
|
usage = SimpleNamespace(
|
|
prompt_tokens=prompt_tokens, completion_tokens=output_tokens, total_tokens=prompt_tokens + output_tokens,
|
|
cache_read_input_tokens=cache_read_tokens, cache_creation_input_tokens=cache_write_tokens,
|
|
)
|
|
finish_reason = _STOP_REASON_TO_FINISH_REASON.get(stop_reason, "stop")
|
|
if self.tool_calls and finish_reason == "stop":
|
|
finish_reason = "tool_calls"
|
|
return SimpleNamespace(
|
|
choices=[SimpleNamespace(index=0, message=msg, finish_reason=finish_reason)], usage=usage, model=model,
|
|
)
|
|
|
|
|
|
def normalize_converse_response(response: Dict) -> SimpleNamespace:
|
|
"""Bedrock Converse response → OpenAI ``ChatCompletion``-shaped SimpleNamespace (``.choices[0].message.
|
|
{content,tool_calls,reasoning_content,reasoning_details,bedrock_content_blocks}``, ``finish_reason``, ``.usage``)."""
|
|
parts = _ResponseParts()
|
|
ordered_blocks = []
|
|
for block in response.get("output", {}).get("message", {}).get("content", []):
|
|
if "text" in block:
|
|
parts.text_parts.append(block["text"])
|
|
ordered_blocks.append({"text": block["text"]})
|
|
elif "reasoningContent" in block:
|
|
ordered_reasoning: Dict[str, Any] = {}
|
|
parts.absorb_reasoning(block["reasoningContent"], ordered_reasoning)
|
|
if ordered_reasoning:
|
|
ordered_blocks.append({"reasoningContent": ordered_reasoning})
|
|
elif "toolUse" in block:
|
|
tu = block["toolUse"]
|
|
ordered_blocks.append(_tool_use_block(tu.get("toolUseId", ""), tu.get("name", ""), tu.get("input", {})))
|
|
parts.tool_calls.append(_tool_call_ns(tu.get("toolUseId", ""), tu.get("name", ""), tu.get("input", {})))
|
|
return parts.build(
|
|
ordered_blocks, response.get("usage", {}), response.get("stopReason", "end_turn"), response.get("modelId", ""),
|
|
)
|
|
|
|
|
|
# --- Streaming response conversion ---
|
|
|
|
def normalize_converse_stream_events(event_stream) -> SimpleNamespace:
|
|
"""Consume a ConverseStream event stream (no callbacks) → same shape as ``normalize_converse_response()``."""
|
|
return stream_converse_with_callbacks(event_stream)
|
|
|
|
|
|
def stream_converse_with_callbacks(
|
|
event_stream, on_text_delta=None, on_tool_start=None, on_reasoning_delta=None,
|
|
on_interrupt_check=None, on_event=None,
|
|
) -> SimpleNamespace:
|
|
"""boto3 ``converse_stream()`` response + callbacks → the ``normalize_converse_response()`` shape.
|
|
``on_text_delta`` only fires while no toolUse block has been seen (as on the Anthropic/chat_completions
|
|
paths); ``on_interrupt_check`` True stops streaming; ``on_event`` fires for EVERY event before branching
|
|
and its exceptions are swallowed so a watchdog hook can never abort the stream.
|
|
|
|
Blocks are keyed by the ``contentBlockIndex`` Bedrock stamps on every contentBlockStart/Delta/Stop:
|
|
text blocks get NO contentBlockStart, so a counter keyed on starts shredded them (#108200)."""
|
|
parts = _ResponseParts()
|
|
stream_blocks: Dict[int, Dict[str, Any]] = {}
|
|
current_block_index: Optional[int] = None
|
|
current_tool: Optional[Dict] = None
|
|
current_text_buffer: List[str] = []
|
|
has_tool_use = False
|
|
stop_reason = None
|
|
interrupted = False
|
|
usage_data: Dict[str, int] = {}
|
|
|
|
def block_index(payload: Dict[str, Any], *, new_block: bool = False) -> int:
|
|
"""Index of the block a contentBlock* event addresses. Without ``contentBlockIndex`` (test doubles,
|
|
proxies) a start opens a fresh slot and a delta/stop continues the current one."""
|
|
idx = payload.get("contentBlockIndex")
|
|
if isinstance(idx, int):
|
|
return idx
|
|
return len(stream_blocks) if new_block or current_block_index is None else current_block_index
|
|
|
|
def flush_text() -> None:
|
|
if current_text_buffer:
|
|
parts.text_parts.append("".join(current_text_buffer))
|
|
current_text_buffer.clear()
|
|
|
|
for event in event_stream.get("stream", []):
|
|
if on_event is not None:
|
|
with suppress(Exception):
|
|
on_event()
|
|
if on_interrupt_check and on_interrupt_check():
|
|
interrupted = True
|
|
break
|
|
if "contentBlockStart" in event:
|
|
start_event = event["contentBlockStart"]
|
|
idx = current_block_index = block_index(start_event, new_block=True)
|
|
start = start_event.get("start", {})
|
|
if "toolUse" in start:
|
|
has_tool_use = True
|
|
flush_text()
|
|
current_tool = {"toolUseId": start["toolUse"].get("toolUseId", ""), "name": start["toolUse"].get("name", ""), "input_json": ""}
|
|
stream_blocks[idx] = _tool_use_block(current_tool["toolUseId"], current_tool["name"], {})
|
|
if on_tool_start:
|
|
on_tool_start(current_tool["name"])
|
|
elif "contentBlockDelta" in event:
|
|
delta_event = event["contentBlockDelta"]
|
|
idx = current_block_index = block_index(delta_event)
|
|
delta = delta_event.get("delta", {})
|
|
if "text" in delta:
|
|
text = delta["text"]
|
|
block = stream_blocks.setdefault(idx, {"text": ""})
|
|
block["text"] = block.get("text", "") + text
|
|
current_text_buffer.append(text)
|
|
if on_text_delta and not has_tool_use:
|
|
on_text_delta(text)
|
|
elif "toolUse" in delta and current_tool is not None:
|
|
current_tool["input_json"] += delta["toolUse"].get("input", "")
|
|
elif "reasoningContent" in delta:
|
|
reasoning = delta["reasoningContent"]
|
|
if isinstance(reasoning, dict) and (reasoning.get("text", "") or reasoning.get("signature") or _encode_redacted(reasoning.get("redactedContent"))):
|
|
block = stream_blocks.setdefault(idx, {"reasoningContent": {}}).setdefault("reasoningContent", {})
|
|
parts.absorb_reasoning(reasoning, block, on_reasoning_delta)
|
|
elif "contentBlockStop" in event:
|
|
idx = block_index(event["contentBlockStop"])
|
|
current_block_index = None # a following index-less delta opens a fresh slot, not this one
|
|
if current_tool is not None:
|
|
input_dict = _parse_tool_args(current_tool["input_json"]) # "" → {} via the JSON-error path
|
|
parts.tool_calls.append(_tool_call_ns(current_tool["toolUseId"], current_tool["name"], input_dict))
|
|
if "toolUse" in stream_blocks.get(idx, {}):
|
|
stream_blocks[idx]["toolUse"]["input"] = input_dict
|
|
current_tool = None
|
|
else:
|
|
flush_text()
|
|
elif "messageStop" in event:
|
|
stop_reason = event["messageStop"].get("stopReason", "end_turn")
|
|
elif "metadata" in event:
|
|
meta_usage = event["metadata"].get("usage", {})
|
|
usage_data = {key: meta_usage.get(key, 0) for key in ("inputTokens", "outputTokens", "cacheReadInputTokens", "cacheWriteInputTokens")}
|
|
if stop_reason is None and not interrupted:
|
|
raise EmptyStreamError("Bedrock Converse stream ended before messageStop; response is incomplete")
|
|
flush_text()
|
|
return parts.build([stream_blocks[i] for i in sorted(stream_blocks)], usage_data, stop_reason or "end_turn", "")
|
|
|
|
|
|
# --- High-level API: call Bedrock Converse ---
|
|
|
|
def build_converse_kwargs(
|
|
model: str, messages: List[Dict], tools: Optional[List[Dict]] = None, max_tokens: Optional[int] = 4096,
|
|
temperature: Optional[float] = None, top_p: Optional[float] = None,
|
|
stop_sequences: Optional[List[str]] = None, guardrail_config: Optional[Dict] = None,
|
|
) -> Dict[str, Any]:
|
|
"""Build kwargs for ``bedrock-runtime.converse()`` / ``converse_stream()``. ``max_tokens=None`` omits
|
|
``maxTokens`` (model maximum; default stays 4096). cachePoint markers go on system, tools and the
|
|
second-newest message (survives as the tail grows — mirrors Anthropic system_and_3), each only if the
|
|
model supports caching and Bedrock has not rejected that placement."""
|
|
system_prompt, converse_messages = convert_messages_to_converse(messages)
|
|
cache_at = {p for p in CACHE_POINT_PLACEMENTS if cache_point_allowed(model, p)} if _model_supports_prompt_cache(model) else set()
|
|
inference_config: Dict[str, Any] = {} if max_tokens is None else {"maxTokens": max_tokens}
|
|
kwargs: Dict[str, Any] = {"modelId": model, "messages": converse_messages, "inferenceConfig": inference_config}
|
|
if system_prompt:
|
|
kwargs["system"] = system_prompt + [dict(_CACHE_POINT)] if "system" in cache_at else system_prompt
|
|
from agent.anthropic_adapter import _forbids_sampling_params
|
|
if not _forbids_sampling_params(model) and not _BEDROCK_XAI_GROK_NO_SAMPLING_RE.match(model or ""):
|
|
inference_config.update({k: v for k, v in (("temperature", temperature), ("topP", top_p)) if v is not None})
|
|
if stop_sequences:
|
|
inference_config["stopSequences"] = stop_sequences
|
|
converse_tools = convert_tools_to_converse(tools) if tools else []
|
|
if converse_tools and not _model_supports_tool_use(model):
|
|
# Non-tool-calling models reject toolConfig (ValidationException → retry loop): strip and warn.
|
|
logger.warning(
|
|
"Model %s does not support tool calling — tools stripped. "
|
|
"The agent will operate in text-only mode.", model
|
|
)
|
|
elif converse_tools:
|
|
kwargs["toolConfig"] = {"tools": converse_tools + [dict(_CACHE_POINT)] if "tools" in cache_at else converse_tools}
|
|
if "messages" in cache_at and len(converse_messages) >= 2:
|
|
content = converse_messages[-2].get("content")
|
|
if isinstance(content, list) and content:
|
|
content.append(dict(_CACHE_POINT))
|
|
if guardrail_config:
|
|
kwargs["guardrailConfig"] = guardrail_config
|
|
if not inference_config:
|
|
del kwargs["inferenceConfig"] # optional on the wire; don't send {}
|
|
return kwargs
|
|
|
|
|
|
def call_converse(
|
|
region: str, model: str, messages: List[Dict], tools: Optional[List[Dict]] = None,
|
|
max_tokens: Optional[int] = 4096, temperature: Optional[float] = None, top_p: Optional[float] = None,
|
|
stop_sequences: Optional[List[str]] = None, guardrail_config: Optional[Dict] = None,
|
|
) -> SimpleNamespace:
|
|
"""Non-streaming Converse call → OpenAI-compatible response. Retries once without a rejected cachePoint
|
|
placement; evicts the cached client on stale-connection errors."""
|
|
client = _get_bedrock_runtime_client(region)
|
|
kwargs = build_converse_kwargs(model, messages, tools, max_tokens, temperature, top_p, stop_sequences, guardrail_config)
|
|
try:
|
|
response = client.converse(**kwargs)
|
|
except Exception as exc:
|
|
retry_kwargs = recover_from_cache_point_rejection(exc, kwargs)
|
|
if retry_kwargs is not None:
|
|
return normalize_converse_response(client.converse(**retry_kwargs))
|
|
redacted_retry_kwargs = recover_from_redacted_reasoning_rejection(exc, kwargs)
|
|
if redacted_retry_kwargs is not None:
|
|
return normalize_converse_response(client.converse(**redacted_retry_kwargs))
|
|
if is_stale_connection_error(exc):
|
|
logger.warning(
|
|
"bedrock: stale-connection error on converse(region=%s, model=%s): "
|
|
"%s — evicting cached client so the next call reconnects.", region, model, type(exc).__name__,
|
|
)
|
|
invalidate_runtime_client(region)
|
|
raise
|
|
return normalize_converse_response(response)
|
|
|
|
|
|
# --- Model discovery ---
|
|
|
|
_discovery_cache: Dict[str, Any] = {}
|
|
_DISCOVERY_CACHE_TTL_SECONDS = 3600
|
|
|
|
|
|
def reset_discovery_cache():
|
|
"""Clear the model discovery cache. Used in tests."""
|
|
_discovery_cache.clear()
|
|
|
|
|
|
def _model_entry(model_id: str, name: Any, provider: str, input_mods: list, output_mods: list) -> Dict[str, Any]:
|
|
return {"id": model_id, "name": (name or model_id).strip(), "provider": provider,
|
|
"input_modalities": input_mods, "output_modalities": output_mods, "streaming": True}
|
|
|
|
|
|
def _list_foundation_models(client, filter_set: set, models: List[Dict[str, Any]]) -> None:
|
|
"""Append active, streaming-capable, text-output foundation models (optionally provider-filtered)."""
|
|
for summary in client.list_foundation_models().get("modelSummaries", []):
|
|
model_id = (summary.get("modelId") or "").strip()
|
|
if not model_id:
|
|
continue
|
|
provider_name = summary.get("providerName") or ""
|
|
model_prefix = model_id.split(".")[0].lower() if "." in model_id else ""
|
|
if filter_set and provider_name.lower() not in filter_set and model_prefix not in filter_set:
|
|
continue
|
|
output_mods = summary.get("outputModalities", [])
|
|
if (summary.get("modelLifecycle", {}).get("status", "").upper() != "ACTIVE"
|
|
or not summary.get("responseStreamingSupported", False) or "TEXT" not in output_mods):
|
|
continue
|
|
models.append(_model_entry(
|
|
model_id, summary.get("modelName"), provider_name.strip(), summary.get("inputModalities", []), output_mods,
|
|
))
|
|
|
|
|
|
def _list_inference_profiles(client, filter_set: set, models: List[Dict[str, Any]]) -> None:
|
|
"""Append active cross-region inference profiles whose IDs are not already present (paginated)."""
|
|
profiles, next_token = [], None
|
|
while True:
|
|
response = client.list_inference_profiles(**({"nextToken": next_token} if next_token else {}))
|
|
profiles.extend(response.get("inferenceProfileSummaries", []))
|
|
if not (next_token := response.get("nextToken")):
|
|
break
|
|
seen_ids = {m["id"].lower() for m in models}
|
|
for profile in profiles:
|
|
profile_id = (profile.get("inferenceProfileId") or "").strip()
|
|
if not profile_id or profile.get("status") != "ACTIVE" or profile_id.lower() in seen_ids:
|
|
continue
|
|
if filter_set and not any(
|
|
_extract_provider_from_arn(m.get("modelArn", "")).lower() in filter_set for m in profile.get("models", [])
|
|
):
|
|
continue
|
|
models.append(_model_entry(profile_id, profile.get("inferenceProfileName"), "inference-profile", ["TEXT"], ["TEXT"]))
|
|
seen_ids.add(profile_id.lower())
|
|
|
|
|
|
def discover_bedrock_models(region: str, provider_filter: Optional[List[str]] = None) -> List[Dict[str, Any]]:
|
|
"""Foundation models + inference profiles (cached 1h per region/filter), ``global.`` profiles first then
|
|
by name; [] when the client cannot be built."""
|
|
# The list is account-scoped (whichever credentials the control client signs with), so a routed
|
|
# profile gets its own entry; unscoped keeps the region:filter key byte-for-byte.
|
|
from hermes_constants import get_hermes_home_override, hermes_home_key
|
|
cache_key = f"{region}:{','.join(sorted(provider_filter or []))}"
|
|
if get_hermes_home_override() is not None:
|
|
cache_key = f"{hermes_home_key()}|{cache_key}"
|
|
cached = _discovery_cache.get(cache_key)
|
|
if cached and (time.time() - cached["timestamp"]) < _DISCOVERY_CACHE_TTL_SECONDS:
|
|
return cached["models"]
|
|
try:
|
|
client = _get_bedrock_control_client(region)
|
|
except Exception as e:
|
|
logger.warning("Failed to create Bedrock client for model discovery: %s", e)
|
|
return []
|
|
models: List[Dict[str, Any]] = []
|
|
filter_set = {f.lower() for f in (provider_filter or [])}
|
|
for step, log, message in (
|
|
(_list_foundation_models, logger.warning, "Failed to list Bedrock foundation models: %s"),
|
|
(_list_inference_profiles, logger.debug, "Skipping inference profile discovery: %s"),
|
|
):
|
|
try:
|
|
step(client, filter_set, models)
|
|
except Exception as e:
|
|
log(message, e)
|
|
models.sort(key=lambda m: (0 if m["id"].startswith("global.") else 1, m["name"].lower()))
|
|
_discovery_cache[cache_key] = {"timestamp": time.time(), "models": models}
|
|
return models
|
|
|
|
|
|
def _extract_provider_from_arn(arn: str) -> str:
|
|
"""``arn:aws:bedrock:...:foundation-model/anthropic.claude-v2`` → ``"anthropic"``."""
|
|
match = re.search(r"foundation-model/([^.]+)", arn)
|
|
return match.group(1) if match else ""
|
|
|
|
|
|
# --- Bedrock model context lengths ---
|
|
# Static fallback when the live probe is unavailable (agent/model_metadata.py). Keys match by longest
|
|
# substring, so versioned entries win over the generic "anthropic.claude-opus-4".
|
|
|
|
BEDROCK_CONTEXT_LENGTHS: Dict[str, int] = {
|
|
# https://docs.aws.amazon.com/bedrock/latest/userguide/model-card-xai-grok-4-6.html
|
|
"xai.grok-4.6": 500_000,
|
|
# Anthropic Claude: 1M GA vs 200K. The 1M entries must match agent/model_metadata.py
|
|
# DEFAULT_CONTEXT_LENGTHS or context compresses early — Opus 5 reached that table and not this
|
|
# one, so the offline path resolved 128K for a 1M model (#74263); the pairing is now tested.
|
|
**dict.fromkeys((
|
|
"anthropic.claude-fable-5", "anthropic.claude-fable", "anthropic.claude-sonnet-5", "anthropic.claude-opus-5",
|
|
"anthropic.claude-opus-4-8", "anthropic.claude-opus-4-7",
|
|
"anthropic.claude-opus-4-6", "anthropic.claude-sonnet-4-6",
|
|
), 1_000_000),
|
|
**dict.fromkeys((
|
|
"anthropic.claude-sonnet-4-5", "anthropic.claude-haiku-4-5", "anthropic.claude-opus-4", "anthropic.claude-sonnet-4",
|
|
"anthropic.claude-3-5-sonnet", "anthropic.claude-3-5-haiku", "anthropic.claude-3-opus", "anthropic.claude-3-sonnet",
|
|
"anthropic.claude-3-haiku",
|
|
), 200_000),
|
|
# Amazon Nova
|
|
**dict.fromkeys(("amazon.nova-pro", "amazon.nova-lite"), 300_000), "amazon.nova-micro": 128_000,
|
|
# Meta Llama / Mistral / DeepSeek
|
|
**dict.fromkeys((
|
|
"meta.llama4-maverick", "meta.llama4-scout", "meta.llama3-3-70b-instruct", "mistral.mistral-large", "deepseek.v3",
|
|
), 128_000),
|
|
# OpenAI on Bedrock (Mantle/Responses route): docs.aws.amazon.com/bedrock/latest/userguide/model-cards-openai.html
|
|
**dict.fromkeys(BEDROCK_OPENAI_RESPONSES_MODEL_IDS, 272_000),
|
|
}
|
|
|
|
BEDROCK_DEFAULT_CONTEXT_LENGTH = 128_000 # unknown Bedrock models
|
|
|
|
# Probe padding tiers (tokens): a wildly oversized payload yields an opaque InternalServerException
|
|
# instead of a clean ValidationException.
|
|
_BEDROCK_PROBE_TIERS = (1_300_000, 2_200_000)
|
|
_WORDS_PER_TOKEN = 0.9 # conservative: ensures the padded prompt clears the tier
|
|
|
|
|
|
def probe_bedrock_context_length(model_id: str, region: str) -> Optional[int]:
|
|
"""Discover a model's real context window by provoking a length error — the only authoritative source
|
|
("prompt is too long: 1300032 tokens > 1000000 maximum"); length validation runs before inference so the
|
|
probe costs nothing. An accepted tier is a safe lower bound; None (no creds/network/unparseable) → static table."""
|
|
from agent.model_metadata import parse_context_limit_from_error
|
|
try:
|
|
client = _get_bedrock_runtime_client(region)
|
|
except Exception as exc: # boto3 missing / credential resolution failure
|
|
logger.debug("Bedrock context probe skipped for %s: %s", model_id, exc)
|
|
return None
|
|
last_error = ""
|
|
for tier_tokens in _BEDROCK_PROBE_TIERS:
|
|
oversized = "data " * int(tier_tokens / _WORDS_PER_TOKEN)
|
|
try:
|
|
client.converse(modelId=model_id, messages=[{"role": "user", "content": [{"text": oversized}]}],
|
|
inferenceConfig={"maxTokens": 8})
|
|
logger.debug("Bedrock context probe for %s accepted ~%s-token prompt; "
|
|
"window is at least that", model_id, f"{tier_tokens:,}")
|
|
return tier_tokens
|
|
except Exception as exc:
|
|
last_error = str(exc)
|
|
limit = parse_context_limit_from_error(last_error)
|
|
if limit and limit >= 1024:
|
|
logger.info("Probed Bedrock context window for %s: %s tokens", model_id, f"{limit:,}")
|
|
return limit
|
|
# Opaque server error / auth / throttle at this tier — try the next.
|
|
logger.debug("Bedrock context probe for %s returned no parseable limit: %s", model_id, last_error[:200])
|
|
return None
|
|
|
|
|
|
def get_bedrock_context_length(model_id: str, region: str = "", probe: bool = True) -> int:
|
|
"""Context window: live probe (if ``probe`` and ``region``) → static table → default. The table is fallback
|
|
only: a stale substring match silently caps the window (a 1M Opus pinned to 200K via "opus-4").
|
|
An application-inference-profile ARN is first resolved to the model it wraps (#114476)."""
|
|
profile_arn = model_id if _APPLICATION_PROFILE_ARN_RE.search(model_id) else ""
|
|
if profile_arn:
|
|
model_id = _resolve_inference_profile_model_id(profile_arn, region)
|
|
if probe and region and (probed := probe_bedrock_context_length(model_id, region)):
|
|
return probed
|
|
matches = [key for key in BEDROCK_CONTEXT_LENGTHS if key in model_id.lower()]
|
|
if matches:
|
|
return BEDROCK_CONTEXT_LENGTHS[max(matches, key=len)]
|
|
if profile_arn:
|
|
logger.warning(
|
|
"Bedrock inference profile %s resolved no known model window; using the %s default. "
|
|
"Grant bedrock:GetInferenceProfile or set model.context_length explicitly if the "
|
|
"wrapped model has a larger window.",
|
|
profile_arn,
|
|
f"{BEDROCK_DEFAULT_CONTEXT_LENGTH:,}",
|
|
)
|
|
return BEDROCK_DEFAULT_CONTEXT_LENGTH
|
|
|
|
|
|
# An application-inference-profile ARN (cost-allocation wrapper) carries an opaque id, so the probe
|
|
# error text and the static substring table both miss and the 128k default silently applies
|
|
# (#114476). System-defined `inference-profile/us.anthropic...` ARNs embed the model id and need no
|
|
# lookup. The ARN's own region (field 4) is authoritative for the control-plane call: the runtime
|
|
# region / base_url may differ, and an empty region must not skip the lookup because the
|
|
# production caller (agent/model_metadata.py::_resolve_bedrock_context_length) passes none.
|
|
_APPLICATION_PROFILE_ARN_RE = re.compile(r":application-inference-profile/")
|
|
_ARN_REGION_RE = re.compile(r"^arn:[^:]+:bedrock:([a-z0-9-]+):", re.IGNORECASE)
|
|
_inference_profile_model_cache: Dict[str, str] = {}
|
|
|
|
|
|
def _resolve_inference_profile_model_id(profile_arn: str, region: str = "") -> str:
|
|
"""Application-profile ARN → the wrapped model's ARN (its ``foundation-model/<id>`` tail satisfies the
|
|
static-table substring match); the profile ARN itself when ``bedrock:GetInferenceProfile`` is not
|
|
granted or unavailable, so callers keep the default-window behaviour. Both outcomes are cached per
|
|
process: this runs on every context-length resolution, not once per model."""
|
|
if profile_arn in _inference_profile_model_cache:
|
|
return _inference_profile_model_cache[profile_arn]
|
|
arn_region = _ARN_REGION_RE.match(profile_arn)
|
|
region = (arn_region.group(1) if arn_region else "") or region or resolve_bedrock_region()
|
|
resolved = profile_arn
|
|
try:
|
|
client = _get_bedrock_control_client(region)
|
|
models = client.get_inference_profile(inferenceProfileIdentifier=profile_arn).get("models") or []
|
|
resolved = next((m["modelArn"] for m in models if m.get("modelArn")), profile_arn)
|
|
except Exception as exc: # no boto3 / credentials / GetInferenceProfile not granted
|
|
logger.debug("Inference profile resolution skipped for %s: %s", profile_arn, exc)
|
|
_inference_profile_model_cache[profile_arn] = resolved
|
|
return resolved
|
|
|
|
|
|
# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
|
|
# Names external plugins imported from this module before the Sep 2026 decomposition.
|
|
# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
|
|
# The whole block is removed by reverting the commit that added it.
|
|
|
|
CONTEXT_OVERFLOW_PATTERNS = [
|
|
re.compile(r"ValidationException.*(?:input is too long|max input token|input token.*exceed)", re.IGNORECASE),
|
|
re.compile(r"ValidationException.*(?:exceeds? the (?:maximum|max) (?:number of )?(?:input )?tokens)", re.IGNORECASE),
|
|
re.compile(r"ModelStreamErrorException.*(?:Input is too long|too many input tokens)", re.IGNORECASE),
|
|
]
|
|
|
|
OVERLOAD_PATTERNS = [
|
|
re.compile(r"ModelNotReadyException", re.IGNORECASE),
|
|
re.compile(r"ModelTimeoutException", re.IGNORECASE),
|
|
re.compile(r"InternalServerException", re.IGNORECASE),
|
|
]
|
|
|
|
THROTTLE_PATTERNS = [
|
|
re.compile(r"ThrottlingException", re.IGNORECASE),
|
|
re.compile(r"Too many concurrent requests", re.IGNORECASE),
|
|
re.compile(r"ServiceQuotaExceededException", re.IGNORECASE),
|
|
]
|
|
|
|
def call_converse_stream(
|
|
region: str,
|
|
model: str,
|
|
messages: List[Dict],
|
|
tools: Optional[List[Dict]] = None,
|
|
max_tokens: Optional[int] = 4096,
|
|
temperature: Optional[float] = None,
|
|
top_p: Optional[float] = None,
|
|
stop_sequences: Optional[List[str]] = None,
|
|
guardrail_config: Optional[Dict] = None,
|
|
) -> SimpleNamespace:
|
|
"""Call Bedrock ConverseStream API and return an OpenAI-compatible response.
|
|
|
|
Consumes the full stream and returns the assembled response. For true
|
|
streaming with delta callbacks, use ``iter_converse_stream()`` instead.
|
|
"""
|
|
client = _get_bedrock_runtime_client(region)
|
|
kwargs = build_converse_kwargs(
|
|
model=model,
|
|
messages=messages,
|
|
tools=tools,
|
|
max_tokens=max_tokens,
|
|
temperature=temperature,
|
|
top_p=top_p,
|
|
stop_sequences=stop_sequences,
|
|
guardrail_config=guardrail_config,
|
|
)
|
|
|
|
try:
|
|
response = client.converse_stream(**kwargs)
|
|
except Exception as exc:
|
|
retry_kwargs = recover_from_cache_point_rejection(exc, kwargs)
|
|
if retry_kwargs is not None:
|
|
return normalize_converse_stream_events(
|
|
client.converse_stream(**retry_kwargs)
|
|
)
|
|
redacted_retry_kwargs = recover_from_redacted_reasoning_rejection(exc, kwargs)
|
|
if redacted_retry_kwargs is not None:
|
|
return normalize_converse_stream_events(
|
|
client.converse_stream(**redacted_retry_kwargs)
|
|
)
|
|
if is_streaming_access_denied_error(exc):
|
|
# IAM allows bedrock:InvokeModel but not
|
|
# InvokeModelWithResponseStream — permanent for this session.
|
|
# Fall back to the non-streaming converse() path.
|
|
logger.info(
|
|
"bedrock: converse_stream denied by IAM on (region=%s, model=%s) — "
|
|
"falling back to non-streaming converse().",
|
|
region, model,
|
|
)
|
|
return normalize_converse_response(client.converse(**kwargs))
|
|
if is_stale_connection_error(exc):
|
|
logger.warning(
|
|
"bedrock: stale-connection error on converse_stream(region=%s, "
|
|
"model=%s): %s — evicting cached client so the next call reconnects.",
|
|
region, model, type(exc).__name__,
|
|
)
|
|
invalidate_runtime_client(region)
|
|
raise
|
|
return normalize_converse_stream_events(response)
|
|
|
|
def is_context_overflow_error(error_message: str) -> bool:
|
|
"""Return True if the error indicates the input context was too large.
|
|
|
|
When this returns True, the agent should compress context and retry
|
|
rather than treating it as a fatal error.
|
|
"""
|
|
return any(p.search(error_message) for p in CONTEXT_OVERFLOW_PATTERNS)
|
|
|
|
def classify_bedrock_error(error_message: str) -> str:
|
|
"""Classify a Bedrock error for retry/failover decisions.
|
|
|
|
Returns:
|
|
- ``"context_overflow"`` — input too long, compress and retry
|
|
- ``"rate_limit"`` — throttled, backoff and retry
|
|
- ``"overloaded"`` — model temporarily unavailable, retry with delay
|
|
- ``"unknown"`` — unclassified error
|
|
"""
|
|
if is_context_overflow_error(error_message):
|
|
return "context_overflow"
|
|
if any(p.search(error_message) for p in THROTTLE_PATTERNS):
|
|
return "rate_limit"
|
|
if any(p.search(error_message) for p in OVERLOAD_PATTERNS):
|
|
return "overloaded"
|
|
return "unknown"
|
|
# ---- END PLUGIN-COMPAT ----
|