From 216a5538d983c79c90c25fee1676514b5d389c87 Mon Sep 17 00:00:00 2001 From: Ranjith Krishnamoorthy Date: Sun, 13 Sep 2026 15:15:57 -0700 Subject: [PATCH 1/3] feat(agentcore): buyer->seller runtime proxy + AgentCore deploy parity Wire the deployed buyer crew to a deployed seller by ARN, and close the remaining gaps between the ECS and AgentCore runtimes for the buyer. Buyer -> seller integration (AgentCoreSellerProxy): - New OpenDirect-shaped proxy backed by bedrock-agentcore:InvokeAgentRuntime against the seller MCP runtime. Issues MCP tools/call over the Streamable-HTTP transport with the required Accept/contentType/protocol headers and the initialize -> notifications/initialized handshake, threads Mcp-Session-Id for microVM stickiness, and parses SSE-framed results. Maps structured seller products to OpenDirect via the shared wire mapper. - crew_tools: use the proxy for the ARN seller case (was a localhost stub). - New CFN stack infra/aws/agentcore/auth-agentcore.yaml: CFN-owned buyer runtime execution role with a scoped runtime/a4a_aamp_seller_* invoke wildcard, restricted to same-account / trusted cross-account (cross-org uses the gateway + CUSTOM_JWT path). deploy.sh deploys it and passes the role via `agentcore configure --execution-role`. AgentCore deploy parity fixes (deploy.sh / requirements.txt): - Add PYTHONPATH=/app/src so ad_buyer is importable in the runtime image. - Default the LLM + memory models to Bedrock Claude Sonnet (was nova-pro / nova-lite), matching settings.py and the Bedrock migration. - Pin crewai>=1.14.4,<2.0.0 (was ==1.10.1) to match pyproject.toml; the Bedrock fix patch and DealBookingFlow rely on CrewAI >=1.14. Tests: - Unit: tests/unit/agentcore/test_agentcore_seller_proxy.py (11) covering MCP result parsing/mapping, required MCP headers, the initialize handshake ordering + initialize-once, and SSE-framed response parsing. - Integration: tests/integration/agentcore/test_runtime.py adds TestBuyerSellerIntegration (deployed buyer->seller loop asserts real seller inv-* inventory via CloudWatch); raise invoke timeouts for the real crew runtime on Claude. - Fix test_agent_hierarchy temperature assertions to be model-aware: build_llm correctly omits temperature for Claude families that reject it (e.g. claude-sonnet-5), so assert the tuned value only when the configured model accepts temperature. All three buyer runtime integration tests + the buyer->seller loop verified live (real inv-* inventory reaches the buyer through the proxy). --- infra/aws/agentcore/auth-agentcore.yaml | 193 +++++++++ infra/aws/agentcore/deploy.sh | 31 +- infra/aws/agentcore/requirements.txt | 11 +- .../agentcore/agentcore_seller_proxy.py | 383 ++++++++++++++++++ .../interfaces/agentcore/crew_tools.py | 22 +- tests/integration/agentcore/test_runtime.py | 138 ++++++- .../agentcore/test_agentcore_seller_proxy.py | 183 +++++++++ tests/unit/test_agent_hierarchy.py | 46 ++- 8 files changed, 989 insertions(+), 18 deletions(-) create mode 100644 infra/aws/agentcore/auth-agentcore.yaml create mode 100644 src/ad_buyer/interfaces/agentcore/agentcore_seller_proxy.py create mode 100644 tests/unit/agentcore/test_agentcore_seller_proxy.py diff --git a/infra/aws/agentcore/auth-agentcore.yaml b/infra/aws/agentcore/auth-agentcore.yaml new file mode 100644 index 00000000..ba07471a --- /dev/null +++ b/infra/aws/agentcore/auth-agentcore.yaml @@ -0,0 +1,193 @@ +AWSTemplateFormatVersion: "2010-09-09" +Description: > + Auth stack for the AAMP buyer AgentCore deployment. Houses the IAM + execution role the buyer runtime assumes, including a SCOPED grant to + invoke seller AgentCore runtimes for buyer->seller (agent-to-agent) calls. + This "auth" stack is the home for auth-related infra (IAM today; Cognito + + gateway authorizer for the cross-org path live alongside it — see NOTE). + +# ============================================================================= +# NOTE — SECURITY MODEL / WHEN TO USE THIS +# ============================================================================= +# The SellerInvoke permission below lets the buyer runtime call the seller's +# AgentCore runtimes directly via bedrock-agentcore:InvokeAgentRuntime, gated +# by AWS IAM. That is appropriate ONLY for: +# +# * Same-account testing — buyer and seller deployed in one AWS account +# (the demo / integration-test setup), OR +# * Customers whose security standards explicitly ALLOW cross-account IAM +# access between the buyer's and seller's accounts (you would then extend +# the resource ARNs / add a cross-account role trust accordingly). +# +# It is NOT the recommended pattern for CROSS-ORGANIZATION buyer<->seller +# interactions (different companies, no shared IAM trust). For that, front the +# seller runtimes with an **AgentCore Gateway + CUSTOM_JWT authorizer backed by +# Cognito** (or the customer's OIDC IdP): the external buyer authenticates with +# a short-lived OAuth2 JWT bearer token at the gateway edge — no IAM in the +# seller's account required. See the `enterprise-auth-gateway` spec. +# +# In short: IAM (this stack) = same-account / trusted cross-account. +# Gateway + CUSTOM_JWT/Cognito = cross-org, zero shared IAM. +# ============================================================================= + +# ============================================================================= +# Parameters +# ============================================================================= +Parameters: + Environment: + Type: String + Default: staging + AllowedValues: [staging, production] + SellerRuntimeNamePrefix: + Type: String + Default: "a4a_aamp_seller_" + Description: > + Name prefix of the seller AgentCore runtimes this buyer may invoke. The + SellerInvoke policy wildcards on runtime/* so the buyer can reach + the seller's http/mcp/a2a runtimes without hardcoding the CLI-assigned + ARN suffixes (which differ per deploy). Keep this as tight as your naming + allows — it is scoped to this account + region + the seller name prefix, + NOT a blanket bedrock-agentcore:* on all runtimes. Match your seller + deploy --name base (e.g. a4a_aamp_seller_ for a4a_aamp_seller_*_mcp). + SellerAccountId: + Type: String + Default: "" + Description: > + Optional. AWS account id that hosts the seller runtimes. Defaults to THIS + account (same-account model). Set this ONLY for a trusted cross-account + setup your security standards permit; the SellerInvoke resource ARNs then + point at that account. (Cross-ORG should use the gateway path instead.) + +# ============================================================================= +# Conditions +# ============================================================================= +Conditions: + # Use this account for the seller ARNs unless a (trusted) cross-account id is given. + UseThisAccountForSeller: !Equals [!Ref SellerAccountId, ""] + +# ============================================================================= +# Resources +# ============================================================================= +Resources: + + # --------------------------------------------------------------------------- + # Buyer AgentCore runtime execution role + # + # Provisioned here (declarative, least-privilege) instead of letting the + # agentcore CLI auto-create a role, so the seller-invoke grant is owned by + # IaC. Pass this role to `agentcore configure --execution-role `. + # --------------------------------------------------------------------------- + BuyerRuntimeExecutionRole: + Type: AWS::IAM::Role + Properties: + RoleName: !Sub "aamp-buyer-${Environment}-runtime-role" + AssumeRolePolicyDocument: + Version: "2012-10-17" + Statement: + - Effect: Allow + Principal: + Service: bedrock-agentcore.amazonaws.com + Action: sts:AssumeRole + Policies: + # -- Standard AgentCore runtime execution permissions -- + # Mirrors the policy the `agentcore` CLI auto-creates for a runtime + # (ECR image pull, CloudWatch Logs + log-delivery, X-Ray tracing, + # CloudWatch metrics, and AgentCore memory/workload access), so a + # CFN-owned role is a drop-in for the CLI-created one. + - PolicyName: AgentCoreRuntimeExecution + PolicyDocument: + Version: "2012-10-17" + Statement: + - Sid: EcrImagePull + Effect: Allow + Action: + - ecr:BatchGetImage + - ecr:GetDownloadUrlForLayer + - ecr:BatchCheckLayerAvailability + Resource: !Sub "arn:aws:ecr:${AWS::Region}:${AWS::AccountId}:repository/bedrock-agentcore-*" + - Sid: EcrAuth + Effect: Allow + Action: ecr:GetAuthorizationToken + Resource: "*" + - Sid: Logs + Effect: Allow + Action: + - logs:CreateLogGroup + - logs:CreateLogStream + - logs:PutLogEvents + - logs:DescribeLogStreams + - logs:DescribeLogGroups + Resource: !Sub "arn:aws:logs:${AWS::Region}:${AWS::AccountId}:log-group:/aws/bedrock-agentcore/runtimes/*" + - Sid: XRayTracing + Effect: Allow + Action: + - xray:PutTraceSegments + - xray:PutTelemetryRecords + - xray:GetSamplingRules + - xray:GetSamplingTargets + Resource: "*" + - Sid: CloudWatchMetrics + Effect: Allow + Action: cloudwatch:PutMetricData + Resource: "*" + - Sid: AgentCoreMemory + Effect: Allow + Action: + - bedrock-agentcore:CreateEvent + - bedrock-agentcore:GetEvent + - bedrock-agentcore:GetMemory + - bedrock-agentcore:GetMemoryRecord + - bedrock-agentcore:ListActors + - bedrock-agentcore:ListEvents + - bedrock-agentcore:ListMemoryRecords + - bedrock-agentcore:ListSessions + Resource: "*" + # -- Bedrock model access (Claude via the Anthropic Messages endpoint) -- + - PolicyName: BedrockInvoke + PolicyDocument: + Version: "2012-10-17" + Statement: + - Effect: Allow + Action: + - bedrock:InvokeModel + - bedrock:InvokeModelWithResponseStream + Resource: "*" + # -- SCOPED seller invoke (see NOTE at top of this file) -- + - PolicyName: SellerInvoke + PolicyDocument: + Version: "2012-10-17" + Statement: + - Sid: InvokeSellerRuntimes + Effect: Allow + Action: bedrock-agentcore:InvokeAgentRuntime + # Wildcard on the seller name prefix + its endpoint children, + # scoped to a single account + region. Not bedrock-agentcore:*. + Resource: + - !If + - UseThisAccountForSeller + - !Sub "arn:aws:bedrock-agentcore:${AWS::Region}:${AWS::AccountId}:runtime/${SellerRuntimeNamePrefix}*" + - !Sub "arn:aws:bedrock-agentcore:${AWS::Region}:${SellerAccountId}:runtime/${SellerRuntimeNamePrefix}*" + - !If + - UseThisAccountForSeller + - !Sub "arn:aws:bedrock-agentcore:${AWS::Region}:${AWS::AccountId}:runtime/${SellerRuntimeNamePrefix}*/*" + - !Sub "arn:aws:bedrock-agentcore:${AWS::Region}:${SellerAccountId}:runtime/${SellerRuntimeNamePrefix}*/*" + Tags: + - Key: Name + Value: !Sub "aamp-buyer-${Environment}-runtime-role" + - Key: Project + Value: ad-buyer-system + - Key: Environment + Value: !Ref Environment + +# ============================================================================= +# Outputs +# ============================================================================= +Outputs: + BuyerRuntimeExecutionRoleArn: + Description: > + Execution role ARN for the buyer AgentCore runtime. Pass to + `agentcore configure --execution-role ` so the runtime can invoke + the seller runtimes (same-account / trusted cross-account only). + Value: !GetAtt BuyerRuntimeExecutionRole.Arn + Export: + Name: !Sub "${AWS::StackName}-BuyerRuntimeExecutionRoleArn" diff --git a/infra/aws/agentcore/deploy.sh b/infra/aws/agentcore/deploy.sh index ce69ee4e..f5826a63 100755 --- a/infra/aws/agentcore/deploy.sh +++ b/infra/aws/agentcore/deploy.sh @@ -28,7 +28,7 @@ set -euo pipefail REGION="${AWS_REGION:-us-west-2}" AGENT_NAME="${AGENT_NAME:-}" AWS_PROFILE="${AWS_PROFILE:-}" -LLM_MODEL="${DEFAULT_LLM_MODEL:-bedrock/us.amazon.nova-pro-v1:0}" +LLM_MODEL="${DEFAULT_LLM_MODEL:-bedrock/us.anthropic.claude-sonnet-4-5-20250929-v1:0}" SELLER_AGENT_URL="${SELLER_AGENT_URL:-}" DEPLOY_MODE="http" DO_TEST=false @@ -143,6 +143,31 @@ if [[ "${TEST_ONLY}" == "false" ]]; then rm -f "${_GENERATED_DOCKERFILE}" fi + # ── Auth stack (opt-in): provision the buyer runtime execution role with a + # scoped grant to invoke the seller runtimes, and pass it to agentcore + # configure. Only when SELLER_AGENT_URL is an AgentCore ARN (buyer->seller + # over InvokeAgentRuntime). For a plain HTTP seller URL or no seller, the CLI + # auto-creates the default role as before. See auth-agentcore.yaml for the + # security model (same-account / trusted cross-account only; cross-org uses + # the gateway + CUSTOM_JWT path). + EXECUTION_ROLE_ARN="" + if [[ "${SELLER_AGENT_URL}" == arn:* ]]; then + AUTH_STACK="aamp-buyer-auth-${ENVIRONMENT:-staging}" + echo "" + echo ">>> Deploying auth stack (buyer runtime role + seller-invoke): ${AUTH_STACK}" + aws cloudformation deploy \ + --stack-name "${AUTH_STACK}" \ + --template-file infra/aws/agentcore/auth-agentcore.yaml \ + --capabilities CAPABILITY_NAMED_IAM \ + --parameter-overrides "Environment=${ENVIRONMENT:-staging}" \ + --region "${REGION}" ${AWS_PROFILE:+--profile "${AWS_PROFILE}"} + EXECUTION_ROLE_ARN=$(aws cloudformation describe-stacks \ + --stack-name "${AUTH_STACK}" --region "${REGION}" ${AWS_PROFILE:+--profile "${AWS_PROFILE}"} \ + --query "Stacks[0].Outputs[?OutputKey=='BuyerRuntimeExecutionRoleArn'].OutputValue" \ + --output text 2>/dev/null || true) + echo " execution role: ${EXECUTION_ROLE_ARN:-}" + fi + # Configure echo "" echo ">>> Configuring agent..." @@ -152,6 +177,7 @@ if [[ "${TEST_ONLY}" == "false" ]]; then -rf infra/aws/agentcore/requirements.txt \ -p HTTP \ -r "${REGION}" \ + ${EXECUTION_ROLE_ARN:+--execution-role "${EXECUTION_ROLE_ARN}"} \ --non-interactive \ --deployment-type container @@ -161,6 +187,7 @@ if [[ "${TEST_ONLY}" == "false" ]]; then agentcore deploy \ --env "DEFAULT_LLM_MODEL=${LLM_MODEL}" \ --env "MANAGER_LLM_MODEL=${LLM_MODEL}" \ + --env "PYTHONPATH=/app/src" \ --env "STORAGE_TYPE=sqlite" \ --env "DATABASE_URL=sqlite:///:memory:" \ --env "ANTHROPIC_API_KEY=not-used-with-bedrock" \ @@ -168,7 +195,7 @@ if [[ "${TEST_ONLY}" == "false" ]]; then --env "AWS_REGION=${REGION}" \ --env "AWS_DEFAULT_REGION=${REGION}" \ --env "CREW_MEMORY_ENABLED=true" \ - --env "MEMORY_LLM_MODEL=bedrock/us.amazon.nova-lite-v1:0" \ + --env "MEMORY_LLM_MODEL=${MEMORY_LLM_MODEL:-bedrock/us.anthropic.claude-sonnet-4-5-20250929-v1:0}" \ --auto-update-on-conflict echo "" diff --git a/infra/aws/agentcore/requirements.txt b/infra/aws/agentcore/requirements.txt index 439b6b8f..ad51e082 100644 --- a/infra/aws/agentcore/requirements.txt +++ b/infra/aws/agentcore/requirements.txt @@ -1,7 +1,16 @@ # Requirements for AgentCore CLI deployment path. # Used by: agentcore configure --requirements-file requirements-agentcore.txt bedrock-agentcore -crewai[tools,anthropic,bedrock]==1.10.1 +# Shared IAB agentic contract catalog. The crew path, deals_client, and the +# AgentCoreSellerProxy import iab_agentic_primitives, so it must be in the +# AgentCore image. Git-free GitHub archive tarball (the CodeBuild image has no +# git). Keep in sync with the seller agent's pin. +iab-agentic-primitives[client] @ https://github.com/IABTechLab/iab-agentic-primitives/archive/refs/tags/v0.5.1.tar.gz +# Pin crewai to match pyproject.toml (>=1.14.4). The Bedrock fix patch +# (patches/crewai_bedrock_fix.py) targets 1.14.x and DealBookingFlow relies on +# CrewAI >=1.14 Flow/state behaviors, so an older pin (the previous ==1.10.1) +# would make the AgentCore image diverge from the ECS/source runtime. +crewai[tools,anthropic,bedrock]>=1.14.4,<2.0.0 anthropic>=0.40.0 httpx>=0.27.0 pydantic>=2.0.0 diff --git a/src/ad_buyer/interfaces/agentcore/agentcore_seller_proxy.py b/src/ad_buyer/interfaces/agentcore/agentcore_seller_proxy.py new file mode 100644 index 00000000..7a422795 --- /dev/null +++ b/src/ad_buyer/interfaces/agentcore/agentcore_seller_proxy.py @@ -0,0 +1,383 @@ +# Author: contributed to IAB Tech Lab (AgentCore adapter) + +"""AgentCore seller proxy — reach a deployed seller runtime by ARN. + +Amazon Bedrock AgentCore exposes a deployed agent as an **ARN**, invoked via +the SigV4-signed ``InvokeAgentRuntime`` data-plane API — NOT as a plain HTTP +URL. The buyer's ``DealBookingFlow`` planning path, however, is written against +the ``OpenDirectClient`` surface (``list_products`` / ``search_products`` / +``get_product`` / ``check_avails``) and expects structured ``Product`` / +``AvailsResponse`` objects. + +``AgentCoreSellerProxy`` bridges the two: it presents the same OpenDirect +method surface the flow needs, but under the hood calls the seller's **MCP** +runtime by ARN via ``bedrock-agentcore:InvokeAgentRuntime``, issuing MCP +``tools/call`` requests (``list_products``, ``get_product_details``, +``discover_inventory``) and mapping the STRUCTURED JSON tool results to the +OpenDirect models at the boundary. + +Why MCP and not A2A for this path: the seller's MCP runtime returns structured +JSON tool results (deterministically parseable into ``Product`` models), +whereas the A2A runtime returns natural-language crew output. A2A is the right +surface for agent-to-agent *negotiation dialogue* (a booking-path concern); the +planning path needs structured product/avails data, so it targets MCP. + +This replaces the previous ``localhost:9999`` stub in ``crew_tools`` for the +deployed (ARN) case. When ``SELLER_AGENT_URL`` is a plain HTTP URL (local dev), +the ordinary ``OpenDirectClient`` is used instead — this proxy is only for the +``arn:`` case. +""" + +from __future__ import annotations + +import json +import logging +from typing import Any + +from ...models.opendirect import AvailsRequest, AvailsResponse, Product +from ...clients.contract_mappers import from_wire_product + +logger = logging.getLogger(__name__) + +# Seller MCP tool names (see seller-agent CREW_MCP_TOOLS / mcp_server). +_TOOL_LIST_PRODUCTS = "list_products" +_TOOL_GET_PRODUCT = "get_product_details" +_TOOL_DISCOVER = "discover_inventory" + + +class AgentCoreSellerProxy: + """OpenDirect-shaped client backed by InvokeAgentRuntime (seller MCP runtime). + + Only the read surface the planning flow exercises is implemented + (``list_products``, ``list_products_tolerant``, ``get_product``, + ``search_products``, ``check_avails``) plus ``close`` and the async context + manager, matching how ``DealBookingFlow`` / the research crews use the + client. Booking (orders/lines/deals) is not on the planning path and is not + proxied here. + + Args: + runtime_arn: The seller **MCP** AgentCore runtime ARN. + region: AWS region of the runtime. + session_id: Optional runtime session id for multi-call continuity. + client: Optional pre-built boto3 ``bedrock-agentcore`` client (tests + inject a stub; production builds one lazily). + """ + + def __init__( + self, + runtime_arn: str, + *, + region: str | None = None, + session_id: str | None = None, + client: Any = None, + ) -> None: + self._arn = runtime_arn + self._region = region + self._session_id = session_id + self._client = client + self._mcp_id = 0 + self._mcp_session_id: str | None = None + self._initialized = False + + # -- boto3 client (lazy) --------------------------------------------------- + + def _get_client(self) -> Any: + if self._client is None: + import boto3 # local import: only needed on the ARN path + + self._client = boto3.client("bedrock-agentcore", region_name=self._region) + return self._client + + # -- InvokeAgentRuntime + MCP framing -------------------------------------- + + def _next_id(self) -> int: + self._mcp_id += 1 + return self._mcp_id + + # AgentCore MCP runtimes speak Streamable-HTTP: the /mcp path is a + # pass-through of the InvokeAgentRuntime payload, but the MCP transport + # REQUIRES the dual Accept header and a JSON content type, and expects the + # standard initialize -> notifications/initialized handshake before any + # tools/call. boto3 exposes these as first-class header params (accept, + # contentType, mcpProtocolVersion, mcpSessionId). Omitting the Accept + # header yields HTTP 406 (McpRequestUnacceptableException, -32011). + _MCP_ACCEPT = "application/json, text/event-stream" + _MCP_CONTENT_TYPE = "application/json" + _MCP_PROTOCOL_VERSION = "2025-03-26" + + def _invoke_mcp(self, payload: dict[str, Any], *, expect_body: bool = True) -> bytes: + """One SigV4 InvokeAgentRuntime call carrying an MCP JSON-RPC message. + + Sets the MCP-required headers, threads the runtime + MCP session ids for + microVM stickiness, and captures the platform-returned ``Mcp-Session-Id`` + for reuse on subsequent calls. Returns the raw response body (empty for + notifications, which get no JSON-RPC reply). + """ + kwargs: dict[str, Any] = { + "agentRuntimeArn": self._arn, + "payload": json.dumps(payload).encode("utf-8"), + "accept": self._MCP_ACCEPT, + "contentType": self._MCP_CONTENT_TYPE, + "mcpProtocolVersion": self._MCP_PROTOCOL_VERSION, + } + if self._session_id: + kwargs["runtimeSessionId"] = self._session_id + if self._mcp_session_id: + kwargs["mcpSessionId"] = self._mcp_session_id + + resp = self._get_client().invoke_agent_runtime(**kwargs) + + # Capture the platform-issued MCP session id (header) for affinity. + sid = (resp.get("mcpSessionId") if isinstance(resp, dict) else None) or ( + resp.get("ResponseMetadata", {}) + .get("HTTPHeaders", {}) + .get("mcp-session-id") + if isinstance(resp, dict) + else None + ) + if sid: + self._mcp_session_id = sid + + if not expect_body: + return b"" + body = resp["response"] + return body.read() if hasattr(body, "read") else bytes(body) + + def _do_initialize(self) -> None: + """Run the MCP initialize + notifications/initialized handshake (sync).""" + init_payload = { + "jsonrpc": "2.0", + "id": self._next_id(), + "method": "initialize", + "params": { + "protocolVersion": self._MCP_PROTOCOL_VERSION, + "capabilities": {}, + "clientInfo": {"name": "ad_buyer.AgentCoreSellerProxy", "version": "1"}, + }, + } + raw = self._invoke_mcp(init_payload) + # Surface a hard protocol/auth error early (406/403 etc. arrive as text). + _raise_for_mcp_error(raw) + # Ack — a notification has no id and expects no body. + self._invoke_mcp( + {"jsonrpc": "2.0", "method": "notifications/initialized", "params": {}}, + expect_body=False, + ) + self._initialized = True + + async def _call_tool(self, name: str, arguments: dict[str, Any] | None = None) -> Any: + """Invoke one MCP ``tools/call`` on the seller runtime, return parsed data. + + Ensures the MCP session is initialized, then issues the ``tools/call`` + with the required MCP headers; the streamed response body is parsed back + into the tool's structured result. + """ + import asyncio + + payload = { + "jsonrpc": "2.0", + "id": self._next_id(), + "method": "tools/call", + "params": {"name": name, "arguments": arguments or {}}, + } + + def _invoke() -> bytes: + if not self._initialized: + self._do_initialize() + return self._invoke_mcp(payload) + + raw = await asyncio.get_event_loop().run_in_executor(None, _invoke) + return _parse_mcp_tool_result(raw) + + # -- OpenDirect surface ---------------------------------------------------- + + async def list_products(self, skip: int = 0, top: int = 50, **filters: Any) -> list[Product]: + data = await self._call_tool(_TOOL_LIST_PRODUCTS, {"limit": top}) + products = _extract_products(data) + if filters: + products = _apply_filters(products, filters) + return products[:top] + + async def list_products_tolerant( + self, skip: int = 0, top: int = 50, **filters: Any + ) -> tuple[list[Product], list[dict[str, Any]]]: + """Tolerant variant: per-product mapping, collecting rejects. + + Mirrors ``OpenDirectClient.list_products_tolerant`` so the per-seller + catalog-resolution path (which prefers the tolerant call) works + unchanged over the proxy. + """ + data = await self._call_tool(_TOOL_LIST_PRODUCTS, {"limit": top}) + raw_items = data.get("products", []) if isinstance(data, dict) else [] + products: list[Product] = [] + rejects: list[dict[str, Any]] = [] + for item in raw_items: + try: + products.append(_map_product(item)) + except Exception as exc: # noqa: BLE001 — collect, don't fail the catalog + rejects.append( + { + "product_id": item.get("product_id") if isinstance(item, dict) else None, + "name": item.get("name") if isinstance(item, dict) else None, + "reason": str(exc), + } + ) + if filters: + products = _apply_filters(products, filters) + return products[:top], rejects + + async def get_product(self, product_id: str) -> Product: + data = await self._call_tool(_TOOL_GET_PRODUCT, {"product_id": product_id}) + # get_product_details may return a bare product or {"product": {...}}. + item = data.get("product", data) if isinstance(data, dict) else data + return _map_product(item) + + async def search_products(self, filters: dict[str, Any]) -> list[Product]: + data = await self._call_tool(_TOOL_LIST_PRODUCTS, {"limit": 500}) + return _apply_filters(_extract_products(data), filters or {}) + + async def check_avails(self, request: AvailsRequest) -> AvailsResponse: + product_id = getattr(request, "product_id", None) or getattr(request, "productId", None) + data = await self._call_tool( + _TOOL_DISCOVER, + {"product_id": product_id} if product_id else {}, + ) + return AvailsResponse.model_validate(_avails_payload(data)) + + async def close(self) -> None: # noqa: D401 — parity with OpenDirectClient + """No persistent connection to close (boto3 client is stateless here).""" + return None + + async def __aenter__(self) -> "AgentCoreSellerProxy": + return self + + async def __aexit__(self, *args: Any) -> None: + await self.close() + + +# -- module-level parsing helpers (unit-testable without AWS) ------------------ + + +def _decode_jsonrpc_body(raw: bytes | str) -> dict | list | None: + """Decode an MCP response body that may be raw JSON or SSE-framed. + + AgentCore MCP runtimes reply with ``Content-Type: text/event-stream`` (we + send ``Accept: application/json, text/event-stream``), so the JSON-RPC + envelope arrives as Server-Sent Events frames:: + + event: message + data: {"jsonrpc":"2.0","id":1,"result":{...}} + + This extracts and parses the ``data:`` payload. A plain-JSON body (no SSE + framing) is parsed directly. Returns the parsed envelope, or ``None`` if the + body is empty / unparseable (callers decide whether that is an error). + """ + if not raw: + return None + text = raw.decode("utf-8") if isinstance(raw, bytes) else raw + stripped = text.lstrip() + # Fast path: already JSON. + if stripped[:1] in "{[": + try: + return json.loads(stripped) + except ValueError: + pass + # SSE framing: concatenate all `data:` lines (last complete JSON wins). + data_lines = [ + line[len("data:") :].strip() + for line in text.splitlines() + if line.startswith("data:") + ] + for chunk in reversed(data_lines): + try: + return json.loads(chunk) + except ValueError: + continue + return None + + +def _raise_for_mcp_error(raw: bytes) -> None: + """Raise if a raw MCP response body carries a JSON-RPC ``error`` object. + + AgentCore returns most MCP errors as HTTP 200 with a JSON-RPC error in the + body (e.g. -32011 Accept-header/406, -32603 internal). boto3 raises for the + genuine HTTP-level 4xx/5xx itself; this catches the in-body errors so the + handshake fails loudly instead of proceeding against a dead session. + """ + env = _decode_jsonrpc_body(raw) + if isinstance(env, dict) and env.get("error"): + raise ValueError(f"seller MCP protocol error: {env['error']}") + + +def _parse_mcp_tool_result(raw: bytes) -> Any: + """Parse an MCP ``tools/call`` response body into the tool's structured data. + + The MCP result shape is ``{"result": {"content": [{"type": "text", + "text": ""}], "isError": false}}``. The tool's real payload is the + JSON string inside the first text content block. Also tolerates a bare + JSON-RPC ``{"result": }`` or a plain object. The transport body may be + raw JSON or SSE-framed (see ``_decode_jsonrpc_body``). + """ + envelope = _decode_jsonrpc_body(raw) + if envelope is None: + raw_txt = (raw.decode("utf-8", "replace") if isinstance(raw, bytes) else str(raw))[:200] + raise ValueError(f"seller MCP response was not JSON/SSE: {raw_txt!r}") + + if isinstance(envelope, dict) and envelope.get("error"): + raise ValueError(f"seller MCP error: {envelope['error']}") + + result = envelope.get("result", envelope) if isinstance(envelope, dict) else envelope + # MCP CallToolResult: content[0].text holds the tool's JSON payload. + if isinstance(result, dict) and isinstance(result.get("content"), list): + for block in result["content"]: + if isinstance(block, dict) and block.get("type") == "text" and block.get("text"): + try: + return json.loads(block["text"]) + except ValueError: + return {"raw_text": block["text"]} + return result + + +def _extract_products(data: Any) -> list[Product]: + items = data.get("products", []) if isinstance(data, dict) else [] + out: list[Product] = [] + for item in items: + try: + out.append(_map_product(item)) + except Exception as exc: # noqa: BLE001 + logger.warning("skipping unmappable product from seller MCP: %s", exc) + return out + + +def _map_product(item: dict[str, Any]) -> Product: + """Map a seller MCP product dict to the OpenDirect ``Product`` model. + + The seller MCP tools emit shared ``iab_agentic_primitives`` Product-shaped + records, so the shared-wire mapper (``from_wire_product``) is the source of + truth — the same boundary mapping ``OpenDirectClient`` uses. A record that + does not validate as a shared Product raises; callers on the tolerant path + (``list_products_tolerant`` / ``_extract_products``) collect the reject + rather than failing the whole catalog. + """ + from iab_agentic_primitives.primitives import Product as WireProduct + + return from_wire_product(WireProduct.model_validate(item)) + + +def _apply_filters(products: list[Product], filters: dict[str, Any]) -> list[Product]: + """Client-side filtering (parity with OpenDirectClient's client-side filter).""" + if not filters: + return products + out = products + inv = filters.get("inventory_type") or filters.get("channel") + if inv: + out = [p for p in out if getattr(p, "inventory_type", None) == inv] + return out + + +def _avails_payload(data: Any) -> dict[str, Any]: + """Normalize a discover_inventory result into an AvailsResponse dict.""" + if isinstance(data, dict) and "available" in data: + return data + # Minimal shape when the discover tool returns a summary/list. + return {"available": True, "products": data if isinstance(data, list) else []} diff --git a/src/ad_buyer/interfaces/agentcore/crew_tools.py b/src/ad_buyer/interfaces/agentcore/crew_tools.py index 72adf6e0..013cff71 100644 --- a/src/ad_buyer/interfaces/agentcore/crew_tools.py +++ b/src/ad_buyer/interfaces/agentcore/crew_tools.py @@ -124,10 +124,26 @@ def run_campaign_plan(prompt: str, brief: dict[str, Any] | None = None) -> dict[ logger.info("Campaign brief: %s", json.dumps(brief, default=str)) - # Create client — dummy URL since seller is a separate AgentCore runtime + # Seller client selection: + # - ARN (deployed seller AgentCore runtime): route through the + # AgentCoreSellerProxy, which calls the seller MCP runtime by ARN via + # InvokeAgentRuntime and returns structured products/avails. Prefer an + # explicit SELLER_MCP_RUNTIME_ARN (the structured surface); otherwise + # fall back to SELLER_AGENT_URL if it is itself an MCP runtime ARN. + # - Plain HTTP URL (local dev): the ordinary OpenDirectClient. seller_url = os.environ.get("SELLER_AGENT_URL", "http://localhost:8001") - client_url = "http://localhost:9999" if seller_url.startswith("arn:") else seller_url - client = OpenDirectClient(base_url=client_url) + mcp_arn = os.environ.get("SELLER_MCP_RUNTIME_ARN", "") + if mcp_arn.startswith("arn:") or seller_url.startswith("arn:"): + from ad_buyer.interfaces.agentcore.agentcore_seller_proxy import ( + AgentCoreSellerProxy, + ) + + client = AgentCoreSellerProxy( + runtime_arn=mcp_arn or seller_url, + region=os.environ.get("AWS_REGION"), + ) + else: + client = OpenDirectClient(base_url=seller_url) flow = DealBookingFlow(client) try: diff --git a/tests/integration/agentcore/test_runtime.py b/tests/integration/agentcore/test_runtime.py index 864c2cb2..132b823b 100644 --- a/tests/integration/agentcore/test_runtime.py +++ b/tests/integration/agentcore/test_runtime.py @@ -80,11 +80,18 @@ def runtime_config(request) -> RuntimeConfig: def invoke_runtime( config: RuntimeConfig, payload: dict, - timeout: int = 120, + timeout: int = 240, max_retries: int = 3, retry_wait: int = 30, ) -> dict: - """Invoke the runtime and return parsed response.""" + """Invoke the runtime and return parsed response. + + Default ``timeout`` is generous: these live tests drive the REAL + multi-agent crew on Bedrock Claude (research + memory save/retrieve + + seller tool calls), which routinely runs ~3 min for a full campaign plan, + plus a possible cold-start on the first invoke after a deploy. Crew/ + integration tests raise it further. + """ payload_json = json.dumps(payload) cmd = ["agentcore", "invoke", payload_json] env = os.environ.copy() @@ -226,3 +233,130 @@ def test_campaign_plan_returns_metadata(self, runtime_config): assert any(kw in response.lower() for kw in ["budget", "campaign", "plan", "allocation"]), ( f"No plan data: {response[:300]}" ) + + +# --------------------------------------------------------------------------- +# Buyer -> Seller runtime-to-runtime integration (AgentCoreSellerProxy) +# --------------------------------------------------------------------------- + + +@pytest.mark.agentcore +class TestBuyerSellerIntegration: + """End-to-end buyer->seller over InvokeAgentRuntime (the deployed loop). + + This is the ONE test that requires both runtimes deployed. The broader + buyer<->seller *functional* contract — pricing tiers by buyer identity, the + deal request/booking flow, MCP tools, order lifecycle, negotiation — is + already covered OFFLINE by the existing suites (e.g. seller + ``tests/integration/test_deal_flow_e2e.py`` / ``test_mcp_integration.py``, + buyer ``tests/integration/test_deal_booking_flow.py`` / + ``test_auth_session_negotiation.py``). This test adds only what those cannot: + proof the real cross-*runtime* transport works when deployed. + + The buyer crew's research path calls ``search_advertising_products``, which + (when ``SELLER_AGENT_URL`` is an AgentCore ARN) is backed by + ``AgentCoreSellerProxy`` -> ``bedrock-agentcore:InvokeAgentRuntime`` against + the seller's **MCP** runtime, issuing MCP ``tools/call`` over the + Streamable-HTTP transport (initialize handshake + dual Accept header + + SSE-framed result parsing). + + Where the signal lives: the crew's FINAL invoke response is a budget + *allocation plan* (channels/percentages/rationale) — it does NOT echo the + seller's product ids. The proof of the closed loop is in the runtime's + CloudWatch logs: the buyer logs the seller catalog's REAL ``inv-*`` ids + coming back through the proxy. (Note: ``inv-*`` is the seller *catalog* + namespace; the buyer crew's own *recommendation* ids use ``prod-*`` and are + unrelated — so only an ``inv-*`` id proves the live seller catalog.) + + Requires BOTH runtimes deployed and the buyer runtime's execution role + granted ``InvokeAgentRuntime`` on the seller runtime (see + ``infra/aws/agentcore/auth-agentcore.yaml``). Skips cleanly when the buyer + was not deployed against an ARN seller, or when CloudWatch is unreadable. + """ + + _REAL_INV_RE = re.compile(r"inv-[a-z0-9-]+", re.IGNORECASE) + + @staticmethod + def _log_group(config: "RuntimeConfig") -> str | None: + """DEFAULT runtime log group for the buyer, derived from its ARN.""" + # arn:...:runtime/ -> /aws/bedrock-agentcore/runtimes/-DEFAULT + m = re.search(r"runtime/([^/]+)$", config.arn) + return f"/aws/bedrock-agentcore/runtimes/{m.group(1)}-DEFAULT" if m else None + + def _recent_log_text(self, config: "RuntimeConfig", since_ms: int, pattern: str) -> str: + """Return recent buyer log messages matching a CloudWatch filter pattern. + + Skips (not fails) when logs are unreadable — the runtime behaviour is + still exercised by the invoke; only the assertion channel is missing. + """ + try: + import boto3 + except ImportError: # pragma: no cover + pytest.skip("boto3 unavailable to read CloudWatch logs") + + lg = self._log_group(config) + if not lg: + pytest.skip(f"Could not derive log group from ARN {config.arn}") + + session = boto3.Session(profile_name=config.profile) if config.profile else boto3.Session() + logs = session.client("logs", region_name=config.region) + try: + lines: list[str] = [] + paginator = logs.get_paginator("filter_log_events") + for page in paginator.paginate( + logGroupName=lg, + startTime=since_ms, + filterPattern=pattern, + ): + lines.extend(e.get("message", "") for e in page.get("events", [])) + if len(lines) > 500: + break + return "\n".join(lines) + except logs.exceptions.ResourceNotFoundException: + pytest.skip(f"Log group {lg} not found — runtime not deployed?") + except Exception as e: # noqa: BLE001 — logs are the assertion channel, not the SUT + pytest.skip(f"Could not read CloudWatch logs: {e}") + + def test_crew_reaches_seller_inventory(self, runtime_config): + """Buyer crew discovers real seller inv-* inventory via the proxy. + + Invokes the crew, then asserts the buyer's logs surfaced real ``inv-*`` + ids from the seller catalog through the proxy path. ``inv-*`` is the + seller *catalog* id namespace; the buyer's own crew *recommendations* + use a separate ``prod-*`` id namespace (see + ``tests/integration/test_deal_booking_flow.py``), so the two are NOT + interchangeable and only an ``inv-*`` id proves the seller catalog was + reached live through the proxy. + """ + since_ms = int((time.time() - 5) * 1000) + result = invoke_runtime( + runtime_config, + { + "prompt": ( + "Plan a $500K Q4 automotive campaign across CTV and digital" + " video, and discover seller inventory." + ), + "routing_mode": "crew", + }, + timeout=300, + ) + assert result["success"], f"Invoke failed: {result['error']}" + + # Did the buyer exercise the seller-search tool at all? (skip guard) + tool_text = self._recent_log_text( + runtime_config, since_ms, '"search_advertising_products"' + ) + if "search_advertising_products" not in tool_text: + pytest.skip( + "Buyer did not exercise the seller-search tool (not wired to an " + "ARN seller?); deploy with --seller-url to run." + ) + + # The definitive signal: real seller inv-* ids in the logs. CloudWatch + # filter patterns need the hyphen term quoted. + inv_text = self._recent_log_text(runtime_config, since_ms, '"inv-"') + inv_ids = sorted(set(self._REAL_INV_RE.findall(inv_text))) + assert inv_ids, ( + "Buyer's seller search returned no real inv-* ids — the proxy->MCP " + f"path returned no catalog data. Log tail: {inv_text[-500:]}" + ) diff --git a/tests/unit/agentcore/test_agentcore_seller_proxy.py b/tests/unit/agentcore/test_agentcore_seller_proxy.py new file mode 100644 index 00000000..4833aeef --- /dev/null +++ b/tests/unit/agentcore/test_agentcore_seller_proxy.py @@ -0,0 +1,183 @@ +"""Unit tests for AgentCoreSellerProxy (offline — boto3 client is stubbed). + +Verifies the proxy presents the OpenDirect surface backed by a mocked +InvokeAgentRuntime returning MCP tools/call envelopes, and that the MCP +result parsing / product mapping helpers behave. +""" + +import json + +import pytest + +from ad_buyer.interfaces.agentcore.agentcore_seller_proxy import ( + AgentCoreSellerProxy, + _parse_mcp_tool_result, +) + +ARN = "arn:aws:bedrock-agentcore:us-west-2:111122223333:runtime/seller_mcp-abc123" + + +def _wire_product(pid: str, name: str = "Apex CTV") -> dict: + return { + "product_id": pid, + "seller_organization_id": "org-seller-1", + "name": name, + "base_price": {"amount_micros": 45_000_000, "currency": "USD"}, + "pricing_type": "floor", + "pricing_model": "cpm", + } + + +def _mcp_envelope(payload: dict) -> bytes: + """Wrap a tool payload as an MCP CallToolResult JSON-RPC response body.""" + return json.dumps( + { + "jsonrpc": "2.0", + "id": 1, + "result": { + "content": [{"type": "text", "text": json.dumps(payload)}], + "isError": False, + }, + } + ).encode("utf-8") + + +class _StreamBody: + def __init__(self, data: bytes): + self._data = data + + def read(self) -> bytes: + return self._data + + +class _FakeBedrockAgentCore: + """Stub boto3 bedrock-agentcore client capturing the last invoke.""" + + def __init__(self, response_payload: dict): + self._payload = response_payload + self.last_kwargs: dict | None = None + + def invoke_agent_runtime(self, **kwargs): + self.last_kwargs = kwargs + return {"response": _StreamBody(_mcp_envelope(self._payload))} + + +# --------------------------------------------------------------------------- # +# Parsing helpers +# --------------------------------------------------------------------------- # + + +class TestParseMcpToolResult: + def test_unwraps_content_text_json(self): + raw = _mcp_envelope({"products": [{"product_id": "inv-ctv-1"}]}) + data = _parse_mcp_tool_result(raw) + assert data == {"products": [{"product_id": "inv-ctv-1"}]} + + def test_raises_on_jsonrpc_error(self): + raw = json.dumps({"jsonrpc": "2.0", "id": 1, "error": {"message": "boom"}}).encode() + with pytest.raises(ValueError, match="seller MCP error"): + _parse_mcp_tool_result(raw) + + def test_raises_on_non_json(self): + with pytest.raises(ValueError, match="not JSON"): + _parse_mcp_tool_result(b"nope") + + def test_unwraps_sse_framed_body(self): + """AgentCore MCP replies as text/event-stream; parse the data: frame.""" + inner = json.dumps( + { + "jsonrpc": "2.0", + "id": 1, + "result": { + "content": [{"type": "text", "text": json.dumps({"products": [{"product_id": "inv-ctv-1"}]})}], + "isError": False, + }, + } + ) + sse = f"event: message\ndata: {inner}\n\n".encode("utf-8") + data = _parse_mcp_tool_result(sse) + assert data == {"products": [{"product_id": "inv-ctv-1"}]} + + +# --------------------------------------------------------------------------- # +# OpenDirect surface over the mocked runtime +# --------------------------------------------------------------------------- # + + +class _RecordingBedrockAgentCore: + """Stub that records EVERY invoke (to assert the MCP handshake sequence).""" + + def __init__(self, response_payload: dict): + self._payload = response_payload + self.calls: list[dict] = [] + + def invoke_agent_runtime(self, **kwargs): + self.calls.append(kwargs) + return {"response": _StreamBody(_mcp_envelope(self._payload))} + + +class TestMcpTransportContract: + async def test_sets_required_mcp_accept_header(self): + """Regression: omitting the dual Accept header returns HTTP 406 (-32011).""" + fake = _RecordingBedrockAgentCore({"products": []}) + proxy = AgentCoreSellerProxy(ARN, client=fake) + await proxy.list_products() + for call in fake.calls: + assert call["accept"] == "application/json, text/event-stream" + assert call["contentType"] == "application/json" + assert call["mcpProtocolVersion"] == "2025-03-26" + + async def test_initialize_handshake_precedes_tools_call(self): + fake = _RecordingBedrockAgentCore({"products": []}) + proxy = AgentCoreSellerProxy(ARN, client=fake) + await proxy.list_products() + methods = [json.loads(c["payload"])["method"] for c in fake.calls] + assert methods == ["initialize", "notifications/initialized", "tools/call"] + + async def test_initialize_runs_once_across_calls(self): + fake = _RecordingBedrockAgentCore({"products": []}) + proxy = AgentCoreSellerProxy(ARN, client=fake) + await proxy.list_products() + await proxy.list_products() + methods = [json.loads(c["payload"])["method"] for c in fake.calls] + assert methods.count("initialize") == 1 + assert methods.count("tools/call") == 2 + + +class TestListProducts: + async def test_list_products_maps_wire_products(self): + fake = _FakeBedrockAgentCore( + {"products": [_wire_product("inv-ctv-apex-sports-nba"), _wire_product("inv-lin-gnn", "GNN")]} + ) + proxy = AgentCoreSellerProxy(ARN, region="us-west-2", client=fake) + products = await proxy.list_products(top=10) + assert [p.name for p in products] == ["Apex CTV", "GNN"] + # It invoked the seller by ARN via the MCP list_products tool. + payload = json.loads(fake.last_kwargs["payload"]) + assert fake.last_kwargs["agentRuntimeArn"] == ARN + assert payload["method"] == "tools/call" + assert payload["params"]["name"] == "list_products" + + async def test_get_product(self): + fake = _FakeBedrockAgentCore({"product": _wire_product("inv-ctv-apex-series", "Apex Series")}) + proxy = AgentCoreSellerProxy(ARN, client=fake) + product = await proxy.get_product("inv-ctv-apex-series") + assert product.name == "Apex Series" + assert json.loads(fake.last_kwargs["payload"])["params"]["name"] == "get_product_details" + + async def test_list_products_tolerant_collects_rejects(self): + # Second item is missing required fields -> reject, not fatal. + fake = _FakeBedrockAgentCore( + {"products": [_wire_product("inv-ok"), {"name": "broken — no ids"}]} + ) + proxy = AgentCoreSellerProxy(ARN, client=fake) + products, rejects = await proxy.list_products_tolerant(top=10) + assert [p.name for p in products] == ["Apex CTV"] + assert len(rejects) == 1 + assert rejects[0]["name"] == "broken — no ids" + + async def test_session_id_threaded(self): + fake = _FakeBedrockAgentCore({"products": []}) + proxy = AgentCoreSellerProxy(ARN, client=fake, session_id="sess-1") + await proxy.list_products() + assert fake.last_kwargs["runtimeSessionId"] == "sess-1" diff --git a/tests/unit/test_agent_hierarchy.py b/tests/unit/test_agent_hierarchy.py index 3981d459..c228d4d8 100644 --- a/tests/unit/test_agent_hierarchy.py +++ b/tests/unit/test_agent_hierarchy.py @@ -43,6 +43,22 @@ _create_research_tools, _format_audience_context, ) +from ad_buyer.llm import _model_accepts_temperature + + +def _expected_temp(tuned: float) -> float | None: + """Temperature an agent's LLM should carry for the configured model. + + Agents pass a tuned temperature to ``build_llm``, but newer Claude families + (e.g. claude-sonnet-5, claude-opus-4-7+) REJECT sampling params, so + ``build_llm`` omits temperature for them and the resulting ``LLM`` reports + ``temperature is None``. These tests must therefore assert the tuned value + only when the *configured* model accepts temperature, and ``None`` when it + does not — otherwise they encode a pre-Bedrock-migration assumption and + fail on the default AgentCore model. Ties the assertion to the same + predicate the production factory uses. + """ + return tuned if _model_accepts_temperature(settings.manager_llm_model) else None # Agent constructors now honor settings.crew_memory_enabled @@ -220,7 +236,7 @@ def test_uses_manager_llm_model(self): def test_llm_temperature_conservative(self): """Manager agent uses conservative temperature for strategic decisions.""" agent = create_portfolio_manager(verbose=False) - assert agent.llm.temperature == 0.3 + assert agent.llm.temperature == _expected_temp(0.3) # =========================================================================== @@ -263,7 +279,7 @@ def test_uses_default_llm_model(self): def test_llm_temperature(self): agent = create_branding_agent(verbose=False) - assert agent.llm.temperature == 0.5 + assert agent.llm.temperature == _expected_temp(0.5) class TestCTVAgent: @@ -293,7 +309,7 @@ def test_tools_injection(self, mock_tools): def test_llm_temperature(self): agent = create_ctv_agent(verbose=False) - assert agent.llm.temperature == 0.5 + assert agent.llm.temperature == _expected_temp(0.5) class TestMobileAppAgent: @@ -319,7 +335,7 @@ def test_memory_enabled(self): def test_llm_temperature(self): agent = create_mobile_app_agent(verbose=False) - assert agent.llm.temperature == 0.5 + assert agent.llm.temperature == _expected_temp(0.5) class TestPerformanceAgent: @@ -345,7 +361,7 @@ def test_memory_enabled(self): def test_llm_temperature(self): agent = create_performance_agent(verbose=False) - assert agent.llm.temperature == 0.5 + assert agent.llm.temperature == _expected_temp(0.5) class TestBuyerDealSpecialistAgent: @@ -399,7 +415,7 @@ def test_custom_tools(self, mock_tools): def test_llm_temperature(self): agent = create_buyer_deal_specialist_agent(verbose=False) - assert agent.llm.temperature == 0.5 + assert agent.llm.temperature == _expected_temp(0.5) # =========================================================================== @@ -432,7 +448,7 @@ def test_memory_enabled(self): def test_low_temperature(self): """Research agent uses low temperature for factual analysis.""" agent = create_research_agent(verbose=False) - assert agent.llm.temperature == 0.2 + assert agent.llm.temperature == _expected_temp(0.2) def test_tools_injection(self, mock_tools): agent = create_research_agent(tools=mock_tools, verbose=False) @@ -467,7 +483,7 @@ def test_memory_enabled(self): def test_very_low_temperature(self): """Execution agent uses very low temperature -- precision matters.""" agent = create_execution_agent(verbose=False) - assert agent.llm.temperature == 0.1 + assert agent.llm.temperature == _expected_temp(0.1) def test_backstory_mentions_booking_states(self): """Execution agent should know the booking lifecycle states.""" @@ -502,7 +518,7 @@ def test_memory_enabled(self): def test_low_temperature(self): agent = create_reporting_agent(verbose=False) - assert agent.llm.temperature == 0.2 + assert agent.llm.temperature == _expected_temp(0.2) def test_backstory_mentions_key_metrics(self): """Should reference advertising KPIs.""" @@ -555,7 +571,7 @@ def test_backstory_mentions_signal_types(self): def test_llm_temperature(self): """Balanced temperature for strategic audience recommendations.""" agent = create_audience_planner_agent(verbose=False) - assert agent.llm.temperature == 0.3 + assert agent.llm.temperature == _expected_temp(0.3) def test_default_no_tools(self): agent = create_audience_planner_agent(verbose=False) @@ -665,6 +681,16 @@ def test_l3_temperatures_lower_than_l2(self): """L3 operational agents should generally use lower temperatures (more precise).""" l2_temps = [factory(verbose=False).llm.temperature for factory in self.L2_FACTORIES] l3_temps = [factory(verbose=False).llm.temperature for factory in self.L3_FACTORIES] + # On temperature-rejecting Claude families build_llm omits temperature + # (all None), so there is no ordering to assert — the tuned values are + # never sent. Only compare when the configured model carries them. + if any(t is None for t in l2_temps + l3_temps): + import pytest + + pytest.skip( + f"Configured model {settings.manager_llm_model!r} rejects temperature; " + "no tuned values to order." + ) avg_l2 = sum(l2_temps) / len(l2_temps) avg_l3 = sum(l3_temps) / len(l3_temps) assert avg_l3 <= avg_l2, f"L3 avg temp ({avg_l3}) should be <= L2 avg temp ({avg_l2})" From 0ec3deefeb47d8215edb33a5572bd51c67ca684e Mon Sep 17 00:00:00 2001 From: Ranjith Krishnamoorthy Date: Fri, 11 Sep 2026 12:33:10 -0700 Subject: [PATCH 2/3] feat(agentcore): run Claude on Bedrock via the Anthropic Messages endpoint Route CrewAI's LLM calls through Amazon Bedrock's Anthropic-compatible (Messages) endpoint instead of the Converse provider, and retire Amazon Nova as the default model. Motivation ---------- The Converse path required a monkey-patch (patches/crewai_bedrock_fix.py) to strip orphaned toolUse/toolResult blocks. Bedrock's Anthropic Messages endpoint (bedrock-runtime .../anthropic) assembles tool turns itself, so that class of ValidationException does not arise. Nova Pro is being deprecated; current-generation Claude tiers (Sonnet 5, Haiku 4.5) are served on the Messages API. Changes ------- - llm.build_llm: add an Anthropic-Messages-compatible base_url branch, selected via ANTHROPIC_COMPATIBLE_LLM_API_BASE_URL. Takes precedence over the OpenAI-compatible branch so Claude is never misrouted to the OpenAI client (Claude is not served on Bedrock's Chat Completions). - settings: add anthropic_compatible_llm_api_{base_url,key}; default model anthropic/claude-sonnet-4-5 -> anthropic/claude-sonnet-5. - http_main: apply the Converse sanitizer only on the legacy Converse path (bedrock/ model + no Messages base URL); on the Messages path apply the new strict-strip patch instead. - patches/crewai_bedrock_anthropic_fix.py (new): strip the tool "strict" field that CrewAI (>=1.15) emits, which Bedrock's Anthropic endpoint rejects (tools.0.custom.strict: Extra inputs are not permitted). Scoped to Bedrock base URLs only; real Anthropic is unchanged. - agentcore memory patch + crew_tools fallback: Nova -> Claude Haiku 4.5. - infra/aws/agentcore/deploy.sh: source .env; default to Sonnet 5 / Haiku 4.5 via the Messages endpoint; set PYTHONPATH=/app/src so the package resolves under the toolkit's module entrypoint. - infra/aws/agentcore/requirements.txt: align crewai with pyproject (>=1.14.4) and add iab-agentic-primitives via a git-free archive URL (the build image has no git binary). - .env.example: document the Messages-endpoint configuration. Compatibility ------------- Backward-compatible: the Converse provider and its patch remain available for anyone using a bedrock/ Converse model. Removing the Converse patch entirely is deferred to a follow-up change. Testing ------- - Unit + mocked integration: 3462 passed. - Live AgentCore integration (deployed dev runtime): 4/4 passed (chat + three crew-plan invocations). --- .env.example | 22 +++++++ infra/aws/agentcore/deploy.sh | 30 ++++++++- infra/aws/agentcore/requirements.txt | 13 +++- patches/crewai_agentcore_memory.py | 4 +- patches/crewai_bedrock_anthropic_fix.py | 60 +++++++++++++++++ src/ad_buyer/config/settings.py | 27 ++++++-- .../interfaces/agentcore/crew_tools.py | 7 +- .../interfaces/agentcore/http_main.py | 48 ++++++++++---- src/ad_buyer/llm/__init__.py | 25 +++++++ tests/integration/agentcore/test_runtime.py | 7 ++ tests/unit/test_agent_hierarchy.py | 19 ++++++ tests/unit/test_llm.py | 66 ++++++++++++++++++- tests/unit/test_settings_lazy_init.py | 32 +++++++-- 13 files changed, 333 insertions(+), 27 deletions(-) create mode 100644 patches/crewai_bedrock_anthropic_fix.py diff --git a/.env.example b/.env.example index 9bfc77d9..473e9925 100644 --- a/.env.example +++ b/.env.example @@ -43,6 +43,28 @@ LLM_MAX_TOKENS=4096 # MANAGER_LLM_MODEL=bedrock/us.anthropic.claude-opus-4-8 # Note: Opus 4.7+ Bedrock IDs are undated (us.anthropic.claude-opus-4-8); older models keep the dated -v1:0 profile form. See https://docs.aws.amazon.com/bedrock/latest/userguide/model-card-anthropic-claude-opus-4-8.html +# Provider: AWS Bedrock via the Anthropic Messages endpoint (RECOMMENDED for +# Bedrock). Routes Claude through CrewAI's native Anthropic provider against +# Bedrock's /anthropic route, so the Converse toolUse/toolResult sanitizer +# (patches/crewai_bedrock_fix.py) is NOT needed. Only current-generation Claude +# tiers are served on the Messages API (Sonnet 5, Opus 4.7/4.8, Haiku 4.5); +# Sonnet 4.5 is Converse-only. Model ids are the Bedrock model / inference- +# profile ids (no provider prefix). +# ANTHROPIC_COMPATIBLE_LLM_API_BASE_URL=https://bedrock-runtime.us-west-2.amazonaws.com/anthropic +# ANTHROPIC_COMPATIBLE_LLM_API_KEY=your-bedrock-api-key +# DEFAULT_LLM_MODEL=us.anthropic.claude-sonnet-5 +# MANAGER_LLM_MODEL=us.anthropic.claude-sonnet-5 +# MEMORY_LLM_MODEL=us.anthropic.claude-haiku-4-5-20251001-v1:0 +# +# AWS credentials for the deploy CLI (dev account). deploy.sh sources this .env, +# so either an AWS_PROFILE that is logged in, or explicit keys, work: +# AWS_PROFILE=genai +# -- or -- +# AWS_ACCESS_KEY_ID=... +# AWS_SECRET_ACCESS_KEY=... +# AWS_SESSION_TOKEN=... +# AWS_REGION=us-west-2 + # Provider: Custom OpenAI-compatible endpoint (alternative) # For endpoints with no CrewAI native provider prefix of their own — NVIDIA # NIM, Ollama, HuggingFace TGI, vLLM, LM Studio, etc. Set DEFAULT_LLM_MODEL to diff --git a/infra/aws/agentcore/deploy.sh b/infra/aws/agentcore/deploy.sh index f5826a63..723b799d 100755 --- a/infra/aws/agentcore/deploy.sh +++ b/infra/aws/agentcore/deploy.sh @@ -25,10 +25,34 @@ set -euo pipefail +# Load a gitignored .env at repo root if present, so local AWS credentials +# (AWS_ACCESS_KEY_ID / AWS_SECRET_ACCESS_KEY / AWS_SESSION_TOKEN or AWS_PROFILE) +# and the Bedrock endpoint config below can be supplied without exporting them +# by hand. .env is listed in .gitignore and must never be committed. +_SCRIPT_DIR_EARLY="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +_REPO_ROOT_EARLY="$(cd "${_SCRIPT_DIR_EARLY}/../../.." && pwd)" +if [[ -f "${_REPO_ROOT_EARLY}/.env" ]]; then + echo ">>> Loading environment from ${_REPO_ROOT_EARLY}/.env" + set -a + # shellcheck disable=SC1091 + source "${_REPO_ROOT_EARLY}/.env" + set +a +fi + REGION="${AWS_REGION:-us-west-2}" AGENT_NAME="${AGENT_NAME:-}" AWS_PROFILE="${AWS_PROFILE:-}" -LLM_MODEL="${DEFAULT_LLM_MODEL:-bedrock/us.anthropic.claude-sonnet-4-5-20250929-v1:0}" +# Default model: current-generation Claude on Bedrock via the Anthropic +# Messages endpoint (see ANTHROPIC_COMPATIBLE_LLM_API_BASE_URL below). This +# replaces the retired Amazon Nova Pro default. The model id is the Bedrock +# model / inference-profile id passed to the Anthropic provider. +LLM_MODEL="${DEFAULT_LLM_MODEL:-us.anthropic.claude-sonnet-5}" +MEMORY_MODEL="${MEMORY_LLM_MODEL:-us.anthropic.claude-haiku-4-5-20251001-v1:0}" +# Bedrock's Anthropic-compatible (Messages API) base URL + API key. Setting +# these routes Claude through CrewAI's native Anthropic provider against +# Bedrock, so the Converse toolUse/toolResult sanitizer is not applied. +ANTHROPIC_BASE_URL="${ANTHROPIC_COMPATIBLE_LLM_API_BASE_URL:-https://bedrock-runtime.${REGION}.amazonaws.com/anthropic}" +BEDROCK_API_KEY="${ANTHROPIC_COMPATIBLE_LLM_API_KEY:-${AWS_BEARER_TOKEN_BEDROCK:-}}" SELLER_AGENT_URL="${SELLER_AGENT_URL:-}" DEPLOY_MODE="http" DO_TEST=false @@ -188,6 +212,8 @@ if [[ "${TEST_ONLY}" == "false" ]]; then --env "DEFAULT_LLM_MODEL=${LLM_MODEL}" \ --env "MANAGER_LLM_MODEL=${LLM_MODEL}" \ --env "PYTHONPATH=/app/src" \ + --env "ANTHROPIC_COMPATIBLE_LLM_API_BASE_URL=${ANTHROPIC_BASE_URL}" \ + --env "ANTHROPIC_COMPATIBLE_LLM_API_KEY=${BEDROCK_API_KEY}" \ --env "STORAGE_TYPE=sqlite" \ --env "DATABASE_URL=sqlite:///:memory:" \ --env "ANTHROPIC_API_KEY=not-used-with-bedrock" \ @@ -195,7 +221,7 @@ if [[ "${TEST_ONLY}" == "false" ]]; then --env "AWS_REGION=${REGION}" \ --env "AWS_DEFAULT_REGION=${REGION}" \ --env "CREW_MEMORY_ENABLED=true" \ - --env "MEMORY_LLM_MODEL=${MEMORY_LLM_MODEL:-bedrock/us.anthropic.claude-sonnet-4-5-20250929-v1:0}" \ + --env "MEMORY_LLM_MODEL=${MEMORY_MODEL}" \ --auto-update-on-conflict echo "" diff --git a/infra/aws/agentcore/requirements.txt b/infra/aws/agentcore/requirements.txt index ad51e082..8a396906 100644 --- a/infra/aws/agentcore/requirements.txt +++ b/infra/aws/agentcore/requirements.txt @@ -1,5 +1,11 @@ # Requirements for AgentCore CLI deployment path. -# Used by: agentcore configure --requirements-file requirements-agentcore.txt +# Used by: agentcore configure --requirements-file infra/aws/agentcore/requirements.txt +# +# Kept in sync with pyproject.toml [project.dependencies]. The AgentCore +# toolkit-generated Dockerfile installs THIS file (it does not `pip install` +# the ad_buyer package), so every runtime import dependency must be listed +# here — including the git-based shared contract library — or the container +# fails at import with ModuleNotFoundError. bedrock-agentcore # Shared IAB agentic contract catalog. The crew path, deals_client, and the # AgentCoreSellerProxy import iab_agentic_primitives, so it must be in the @@ -24,3 +30,8 @@ sqlalchemy>=2.0.0 aiosqlite>=0.20.0 pyyaml>=6.0.0 nest-asyncio>=1.6.0 +# NOTE: the AgentCore build image (astral uv slim) has no `git` binary, so a +# `git+https://` source fails with "Git executable not found". Use the GitHub +# archive tarball instead — a plain HTTPS download that needs no git. Keep this +# pinned to the same tag as pyproject.toml's git ref (v0.5.1). +iab-agentic-primitives[client] @ https://github.com/IABTechLab/iab-agentic-primitives/archive/refs/tags/v0.5.1.tar.gz diff --git a/patches/crewai_agentcore_memory.py b/patches/crewai_agentcore_memory.py index 320c6e03..abac8009 100644 --- a/patches/crewai_agentcore_memory.py +++ b/patches/crewai_agentcore_memory.py @@ -119,7 +119,9 @@ def _patched_memory_init(self, *args, **kwargs): if "llm" not in kwargs or kwargs.get("llm") == "gpt-4o-mini": bedrock_model = os.environ.get( "MEMORY_LLM_MODEL", - os.environ.get("DEFAULT_LLM_MODEL", "bedrock/us.amazon.nova-lite-v1:0"), + os.environ.get( + "DEFAULT_LLM_MODEL", "bedrock/us.anthropic.claude-haiku-4-5-v1:0" + ), ) kwargs["llm"] = bedrock_model try: diff --git a/patches/crewai_bedrock_anthropic_fix.py b/patches/crewai_bedrock_anthropic_fix.py new file mode 100644 index 00000000..203dc1c3 --- /dev/null +++ b/patches/crewai_bedrock_anthropic_fix.py @@ -0,0 +1,60 @@ +"""Patch for CrewAI AnthropicCompletion — Bedrock Anthropic Messages compat. + +When Claude runs on Amazon Bedrock's Anthropic-compatible (Messages) endpoint +(https://bedrock-runtime..amazonaws.com/anthropic), CrewAI's native +Anthropic provider builds tool definitions that include a ``strict`` field. + +CrewAI (>=1.15) unconditionally sets ``"strict": True`` on every tool's +function schema in ``crewai.utilities.agent_utils`` (build_tools), and +``AnthropicCompletion._convert_tools_for_interference`` copies it onto the +Anthropic tool object. Anthropic's own API accepts ``strict``; Amazon Bedrock's +Anthropic-compatible endpoint does NOT, and rejects the request with: + + 400 invalid_request_error: tools.0.custom.strict: Extra inputs are not permitted + +This patch wraps ``_convert_tools_for_interference`` to remove the ``strict`` +key from the produced tool objects, but ONLY when the completion is pointed at +a Bedrock base URL (so behavior against real Anthropic is unchanged). It is the +Messages-endpoint analog of the retired Converse ``crewai_bedrock_fix`` and is +applied only when ``ANTHROPIC_COMPATIBLE_LLM_API_BASE_URL`` is set. + +Usage: + from patches.crewai_bedrock_anthropic_fix import apply_patches + apply_patches() # call once at startup, before building crews +""" + +import logging + +logger = logging.getLogger(__name__) + +_patched = False + + +def _is_bedrock_base_url(url: object) -> bool: + return isinstance(url, str) and ("bedrock-runtime" in url or "bedrock-mantle" in url) + + +def apply_patches() -> None: + """Strip the unsupported ``strict`` tool field on the Bedrock Messages endpoint.""" + global _patched + if _patched: + return + try: + from crewai.llms.providers.anthropic.completion import AnthropicCompletion + except ImportError: + logger.warning("AnthropicCompletion not available — skipping Bedrock Anthropic patch") + return + + original = AnthropicCompletion._convert_tools_for_interference + + def _patched_convert(self, tools): + result = original(self, tools) + if _is_bedrock_base_url(getattr(self, "base_url", None)) and isinstance(result, list): + for tool in result: + if isinstance(tool, dict): + tool.pop("strict", None) + return result + + AnthropicCompletion._convert_tools_for_interference = _patched_convert + _patched = True + logger.info("Patched AnthropicCompletion._convert_tools_for_interference (Bedrock strict-strip)") diff --git a/src/ad_buyer/config/settings.py b/src/ad_buyer/config/settings.py index a1fe8fe2..3cb4ca85 100644 --- a/src/ad_buyer/config/settings.py +++ b/src/ad_buyer/config/settings.py @@ -169,11 +169,17 @@ def resolve_seller_base_url(self) -> str: # LLM Settings # Supported providers: anthropic (default), openai, gemini, bedrock # Set DEFAULT_LLM_MODEL to switch provider, e.g.: - # anthropic/claude-sonnet-4-5-20250929 (requires ANTHROPIC_API_KEY) - # openai/gpt-4o (requires OPENAI_API_KEY) - # gemini/gemini-2.5-flash (requires GOOGLE_API_KEY) - # bedrock/us.anthropic.claude-sonnet-4-5-20250929-v1:0 (requires AWS creds) - default_llm_model: str = "anthropic/claude-sonnet-4-5-20250929" + # anthropic/claude-sonnet-5 (requires ANTHROPIC_API_KEY) + # openai/gpt-4o (requires OPENAI_API_KEY) + # gemini/gemini-2.5-flash (requires GOOGLE_API_KEY) + # bedrock/us.anthropic.claude-sonnet-4-5-20250929-v1:0 (Converse; requires AWS creds) + # NOTE: to run Claude on Amazon Bedrock WITHOUT the Converse provider, do + # not use the ``bedrock/`` prefix; instead set + # ANTHROPIC_COMPATIBLE_LLM_API_BASE_URL to Bedrock's /anthropic route (see + # below) with a Messages-supported model id. Only the current-generation + # Claude tiers are served on the Anthropic Messages API (Sonnet 5, + # Opus 4.7/4.8, Haiku 4.5); Sonnet 4.5 is Converse-only. + default_llm_model: str = "anthropic/claude-sonnet-5" manager_llm_model: str = "anthropic/claude-opus-4-8" llm_temperature: float = 0.3 llm_max_tokens: int = 4096 @@ -187,6 +193,17 @@ def resolve_seller_base_url(self) -> str: openai_compatible_llm_api_key: str | None = None openai_compatible_llm_api_base_url: str | None = None + # Alternative: any Anthropic-Messages-wire-compatible endpoint. This is the + # recommended way to run Claude on Amazon Bedrock: set + # ANTHROPIC_COMPATIBLE_LLM_API_BASE_URL=https://bedrock-runtime..amazonaws.com/anthropic + # ANTHROPIC_COMPATIBLE_LLM_API_KEY= + # DEFAULT_LLM_MODEL / MANAGER_LLM_MODEL = the Bedrock model / inference-profile id + # CrewAI's native Anthropic provider then drives messages.create against + # Bedrock, so no Bedrock Converse toolUse/toolResult sanitization is needed. + # (Prefer bedrock-runtime over bedrock-mantle for new apps per AWS.) + anthropic_compatible_llm_api_key: str | None = None + anthropic_compatible_llm_api_base_url: str | None = None + # Database / Storage Configuration database_url: str = "sqlite:///./ad_buyer.db" redis_url: str | None = None diff --git a/src/ad_buyer/interfaces/agentcore/crew_tools.py b/src/ad_buyer/interfaces/agentcore/crew_tools.py index 013cff71..5a7f4393 100644 --- a/src/ad_buyer/interfaces/agentcore/crew_tools.py +++ b/src/ad_buyer/interfaces/agentcore/crew_tools.py @@ -96,10 +96,13 @@ def run_campaign_plan(prompt: str, brief: dict[str, Any] | None = None) -> dict[ from ad_buyer.flows.deal_booking_flow import DealBookingFlow from ad_buyer.models.flow_state import BookingState - # Override buyer settings to use Bedrock instead of Anthropic. + # Override buyer settings to use the Bedrock-hosted model. Deploy.sh sets + # DEFAULT_LLM_MODEL plus ANTHROPIC_COMPATIBLE_LLM_API_BASE_URL so this + # routes through the Anthropic Messages endpoint (no Converse patch). The + # fallback is a current-generation, Messages-supported Claude id. bedrock_model = os.environ.get( "DEFAULT_LLM_MODEL", - "bedrock/us.anthropic.claude-sonnet-4-5-20250929-v1:0", + "us.anthropic.claude-sonnet-5-v1:0", ) from ad_buyer.config.settings import settings as buyer_settings diff --git a/src/ad_buyer/interfaces/agentcore/http_main.py b/src/ad_buyer/interfaces/agentcore/http_main.py index 8b156622..44a324e5 100644 --- a/src/ad_buyer/interfaces/agentcore/http_main.py +++ b/src/ad_buyer/interfaces/agentcore/http_main.py @@ -55,17 +55,43 @@ os.environ.setdefault("STORAGE_TYPE", "sqlite") os.environ.setdefault("DATABASE_URL", "sqlite:///:memory:") -# Apply CrewAI patches BEFORE any CrewAI imports -try: - from patches.crewai_bedrock_fix import apply_patches as apply_bedrock_patches - - apply_bedrock_patches() -except ImportError as _ie: - logging.getLogger(__name__).warning( - "crewai_bedrock_fix not found: %s (repo_root=%s)", _ie, _repo_root - ) -except Exception as _exc: - logging.getLogger(__name__).warning("crewai_bedrock_fix patch failed: %s", _exc) +# Apply CrewAI patches BEFORE any CrewAI imports. +# +# The Bedrock *Converse* sanitizer (crewai_bedrock_fix) is only needed on the +# legacy Converse path (DEFAULT_LLM_MODEL="bedrock/..."). The recommended setup +# routes Claude through Bedrock's Anthropic Messages endpoint by setting +# ANTHROPIC_COMPATIBLE_LLM_API_BASE_URL, which uses CrewAI's native Anthropic +# provider (messages.create) and does NOT need Converse tool-block +# sanitization. So we apply the Converse patch only when that base URL is unset +# AND the model is a bedrock/ Converse model. +_default_model = os.environ.get("DEFAULT_LLM_MODEL", "") +_on_converse_path = not os.environ.get( + "ANTHROPIC_COMPATIBLE_LLM_API_BASE_URL" +) and _default_model.startswith("bedrock/") +if _on_converse_path: + try: + from patches.crewai_bedrock_fix import apply_patches as apply_bedrock_patches + + apply_bedrock_patches() + except ImportError as _ie: + logging.getLogger(__name__).warning( + "crewai_bedrock_fix not found: %s (repo_root=%s)", _ie, _repo_root + ) + except Exception as _exc: + logging.getLogger(__name__).warning("crewai_bedrock_fix patch failed: %s", _exc) +else: + # Anthropic Messages endpoint path: CrewAI (>=1.15) emits a ``strict`` tool + # field that Bedrock's Anthropic-compatible endpoint rejects. Strip it. + try: + from patches.crewai_bedrock_anthropic_fix import apply_patches as apply_anthropic_patches + + apply_anthropic_patches() + except ImportError as _ie: + logging.getLogger(__name__).warning( + "crewai_bedrock_anthropic_fix not found: %s (repo_root=%s)", _ie, _repo_root + ) + except Exception as _exc: + logging.getLogger(__name__).warning("crewai_bedrock_anthropic_fix patch failed: %s", _exc) try: from patches.crewai_agentcore_memory import apply_patches as apply_memory_patches diff --git a/src/ad_buyer/llm/__init__.py b/src/ad_buyer/llm/__init__.py index 1144940d..b48b9d6f 100644 --- a/src/ad_buyer/llm/__init__.py +++ b/src/ad_buyer/llm/__init__.py @@ -27,6 +27,11 @@ # that speaks the OpenAI wire format (NVIDIA NIM, Ollama, HuggingFace TGI, ...). _OPENAI_COMPATIBLE_PROVIDER = "openai" +# CrewAI's native Anthropic SDK client; with a base_url it drives any endpoint +# that speaks the Anthropic Messages wire format (Amazon Bedrock's /anthropic +# route, a self-hosted Anthropic-compatible gateway, ...). +_ANTHROPIC_COMPATIBLE_PROVIDER = "anthropic" + # Anthropic removed sampling parameters (temperature/top_p/top_k) starting # with Opus 4.7; sending temperature to these families returns # "400 invalid_request_error: 'temperature' is deprecated for this model." @@ -63,6 +68,26 @@ def build_llm(model: str, temperature: float, max_tokens: int) -> LLM: if _model_accepts_temperature(model): kwargs["temperature"] = temperature + # Amazon Bedrock's Anthropic-compatible endpoint (and any other endpoint + # that speaks the Anthropic Messages wire format). Setting + # ANTHROPIC_COMPATIBLE_LLM_API_BASE_URL routes Claude through CrewAI's + # native Anthropic provider (messages.create) against that base URL, + # instead of the Bedrock Converse provider. This is the path that lets us + # run Claude on Bedrock WITHOUT the Converse toolUse/toolResult sanitizer: + # the Messages API assembles tool turns itself, so the orphaned-block + # ValidationException class does not arise. Use a Messages-supported model + # id (e.g. Claude Sonnet 5 / Opus 4.7+ / Haiku 4.5) and a Bedrock API key. + # For bedrock-runtime, base_url is + # https://bedrock-runtime..amazonaws.com/anthropic + # and the model id is the Bedrock model / inference-profile id. + if settings.anthropic_compatible_llm_api_base_url: + return LLM( + api_key=settings.anthropic_compatible_llm_api_key, + provider=_ANTHROPIC_COMPATIBLE_PROVIDER, + base_url=settings.anthropic_compatible_llm_api_base_url, + **kwargs, + ) + if settings.openai_compatible_llm_api_base_url: return LLM( api_key=settings.openai_compatible_llm_api_key, diff --git a/tests/integration/agentcore/test_runtime.py b/tests/integration/agentcore/test_runtime.py index 132b823b..7193ece2 100644 --- a/tests/integration/agentcore/test_runtime.py +++ b/tests/integration/agentcore/test_runtime.py @@ -196,6 +196,11 @@ def test_campaign_plan_with_budget(self, runtime_config): ), "routing_mode": "crew", }, + # Crew mode fans out a multi-agent crew over multi-turn tool calls on + # Claude Sonnet 5; observed end-to-end latency is ~4 min, well past the + # 120s default. Give real headroom so a correct plan isn't scored as a + # failure purely on latency. + timeout=360, ) assert result["success"], f"Invoke failed: {result['error']}" response = result["response"].lower() @@ -211,6 +216,7 @@ def test_campaign_plan_includes_next_step(self, runtime_config): "prompt": "Plan a $200K Q1 campaign for mobile and display", "routing_mode": "crew", }, + timeout=360, ) assert result["success"], f"Invoke failed: {result['error']}" response = result["response"].lower() @@ -226,6 +232,7 @@ def test_campaign_plan_returns_metadata(self, runtime_config): "prompt": "Plan a $1M Q3 brand awareness campaign", "routing_mode": "crew", }, + timeout=360, ) assert result["success"], f"Invoke failed: {result['error']}" # The raw output should contain campaign plan indicators diff --git a/tests/unit/test_agent_hierarchy.py b/tests/unit/test_agent_hierarchy.py index c228d4d8..5029baf8 100644 --- a/tests/unit/test_agent_hierarchy.py +++ b/tests/unit/test_agent_hierarchy.py @@ -69,6 +69,25 @@ def _force_memory_enabled(monkeypatch): monkeypatch.setattr(settings, "crew_memory_enabled", True) +# The default model is now Claude Sonnet 5, a tier on which Anthropic rejects +# the `temperature` parameter (build_llm therefore omits it, leaving +# llm.temperature is None). This module's assertions are about per-agent +# temperature *wiring* and the L3 Settings: - """Build an isolated Settings instance that ignores any local .env file.""" + """Build an isolated Settings instance that ignores any local .env file. + + Explicitly nulls the custom-endpoint fields unless a test overrides them, + so an ambient repo ``.env`` (or exported ``*_COMPATIBLE_*`` env vars) can't + bleed in and flip which provider branch ``build_llm`` takes. Without this, + a checkout that has a real ``.env`` pointing at the Bedrock Anthropic + endpoint would make the OpenAI-compatible-branch tests see the anthropic + branch instead. + """ overrides.setdefault("anthropic_api_key", "sk-ant-test") + overrides.setdefault("openai_compatible_llm_api_base_url", None) + overrides.setdefault("openai_compatible_llm_api_key", None) + overrides.setdefault("anthropic_compatible_llm_api_base_url", None) + overrides.setdefault("anthropic_compatible_llm_api_key", None) return Settings(_env_file=None, **overrides) @@ -82,6 +94,58 @@ def test_local_ollama_needs_no_key(self, monkeypatch): assert llm.base_url == "http://localhost:11434/v1" +class TestBedrockAnthropicCompatibleEndpoint: + """ANTHROPIC_COMPATIBLE_LLM_API_BASE_URL configured — routes Claude through + CrewAI's native Anthropic provider (Messages API) against a custom base URL + such as Amazon Bedrock's /anthropic endpoint. This is the path that lets + Claude run on Bedrock WITHOUT the Converse toolUse/toolResult sanitizer.""" + + def test_bedrock_messages_routes_via_anthropic_with_base_url(self, monkeypatch): + monkeypatch.setattr( + "ad_buyer.llm.get_settings", + lambda: _settings( + anthropic_compatible_llm_api_key="bedrock-key-test", + anthropic_compatible_llm_api_base_url="https://bedrock-runtime.us-west-2.amazonaws.com/anthropic", + ), + ) + llm = build_llm( + model="us.anthropic.claude-sonnet-5-v1:0", + temperature=0.3, + max_tokens=4096, + ) + assert llm.provider == "anthropic" + assert llm.base_url == "https://bedrock-runtime.us-west-2.amazonaws.com/anthropic" + assert llm.api_key == "bedrock-key-test" + + def test_anthropic_compatible_takes_precedence_over_openai_compatible(self, monkeypatch): + """When both custom endpoints are set, the Anthropic-compatible branch + wins so Claude never gets misrouted to the OpenAI client (Claude is not + served on Bedrock's OpenAI Chat Completions path).""" + monkeypatch.setattr( + "ad_buyer.llm.get_settings", + lambda: _settings( + anthropic_compatible_llm_api_key="bedrock-key-test", + anthropic_compatible_llm_api_base_url="https://bedrock-runtime.us-west-2.amazonaws.com/anthropic", + openai_compatible_llm_api_key="should-not-win", + openai_compatible_llm_api_base_url="https://integrate.api.nvidia.com/v1", + ), + ) + llm = build_llm(model="us.anthropic.claude-sonnet-5-v1:0", temperature=0.3, max_tokens=4096) + assert llm.provider == "anthropic" + assert llm.base_url == "https://bedrock-runtime.us-west-2.amazonaws.com/anthropic" + + def test_sonnet_5_omits_temperature_on_bedrock_messages(self, monkeypatch): + monkeypatch.setattr( + "ad_buyer.llm.get_settings", + lambda: _settings( + anthropic_compatible_llm_api_base_url="https://bedrock-runtime.us-west-2.amazonaws.com/anthropic", + ), + ) + llm = build_llm(model="us.anthropic.claude-sonnet-5-v1:0", temperature=0.3, max_tokens=4096) + assert llm.provider == "anthropic" + assert llm.temperature is None + + class TestModelAcceptsTemperature: """Anthropic rejects the temperature parameter on Opus 4.7+, Sonnet 5+, and Fable/Mythos with a 400 invalid_request_error; older models still diff --git a/tests/unit/test_settings_lazy_init.py b/tests/unit/test_settings_lazy_init.py index f9bf2fc1..22471771 100644 --- a/tests/unit/test_settings_lazy_init.py +++ b/tests/unit/test_settings_lazy_init.py @@ -16,13 +16,37 @@ import sys from unittest.mock import patch +import pytest + +_SETTINGS_MOD = "ad_buyer.config.settings" + + +@pytest.fixture(autouse=True) +def _restore_settings_module(): + """Restore the original settings module after each test in this file. + + The tests below ``del sys.modules[...]`` and re-import the settings module + to exercise lazy init with an empty cache. Without restoration the rebuilt + module leaks into ``sys.modules``, so later tests in the full suite that do + ``from ad_buyer.config.settings import settings`` bind to a fresh lazy + proxy reading live env — which made + ``test_real_model_path_e2e::test_label_reflects_active_mode`` fail only in + suite order. Snapshot and restore the original module object here so the + reload stays contained to this file. + """ + original = sys.modules.get(_SETTINGS_MOD) + try: + yield + finally: + if original is not None: + sys.modules[_SETTINGS_MOD] = original + def _reload_settings_module(): """Force a fresh import of the settings module so its lru_cache is empty.""" - mod_name = "ad_buyer.config.settings" - if mod_name in sys.modules: - del sys.modules[mod_name] - return importlib.import_module(mod_name) + if _SETTINGS_MOD in sys.modules: + del sys.modules[_SETTINGS_MOD] + return importlib.import_module(_SETTINGS_MOD) def test_importing_settings_does_not_construct_eagerly(): From 7932f8bbed034d3dbcd7d69b935535a3aeb874e6 Mon Sep 17 00:00:00 2001 From: Ranjith Krishnamoorthy Date: Fri, 11 Sep 2026 18:02:25 -0700 Subject: [PATCH 3/3] fix(llm): send max_completion_tokens for GPT-5.x / o-series models OpenAI's newer models (GPT-5.x, o1/o3/o4 reasoning models) reject the legacy max_tokens parameter on Bedrock's OpenAI-compatible endpoint with "400 invalid_request_error: Unsupported parameter: 'max_tokens' is not supported with this model." They take max_completion_tokens instead. build_llm now routes the token cap to max_completion_tokens for those model families (matched as substrings so Bedrock inference-profile ids like us.openai.gpt-5.6-sol and raw ids like gpt-5/o1 are covered) and keeps max_tokens for everything else (gpt-4o, Claude, Nova, ...). Surfaced while running the buyer crew against a Bedrock OpenAI (GPT-5.6) model to validate the "any model on Bedrock" path; with the fix the GPT crew invoke completes cleanly. Adds regression tests for the helper and the build_llm parameter routing. --- src/ad_buyer/llm/__init__.py | 30 ++++++++++++++++- tests/unit/test_llm.py | 62 +++++++++++++++++++++++++++++++++++- 2 files changed, 90 insertions(+), 2 deletions(-) diff --git a/src/ad_buyer/llm/__init__.py b/src/ad_buyer/llm/__init__.py index b48b9d6f..d156dbf0 100644 --- a/src/ad_buyer/llm/__init__.py +++ b/src/ad_buyer/llm/__init__.py @@ -55,6 +55,28 @@ def _model_accepts_temperature(model: str) -> bool: return not any(family in model_id for family in _TEMPERATURE_REJECTING_MODEL_FAMILIES) +# OpenAI's newer models (GPT-5.x and the o-series reasoning models) reject the +# legacy ``max_tokens`` parameter — sending it returns +# "400 invalid_request_error: Unsupported parameter: 'max_tokens' is not +# supported with this model." They take ``max_completion_tokens`` instead. +# Matched as substrings so a Bedrock inference-profile id +# ("us.openai.gpt-5.6-sol", ...) or a raw OpenAI id ("gpt-5", "o1", ...) is +# covered. Other models (gpt-4o, gpt-oss, Claude, Nova, ...) keep ``max_tokens``. +_MAX_COMPLETION_TOKENS_MODEL_FAMILIES = ( + "gpt-5", + "gpt-6", + "o1", + "o3", + "o4", +) + + +def _model_uses_max_completion_tokens(model: str) -> bool: + """Return ``True`` if ``model`` needs ``max_completion_tokens`` (not ``max_tokens``).""" + model_id = model.lower() + return any(family in model_id for family in _MAX_COMPLETION_TOKENS_MODEL_FAMILIES) + + def build_llm(model: str, temperature: float, max_tokens: int) -> LLM: """Build an ``LLM`` for ``model``, honoring a custom base URL if configured. @@ -64,7 +86,13 @@ def build_llm(model: str, temperature: float, max_tokens: int) -> LLM: """ settings = get_settings() - kwargs: dict[str, Any] = {"model": model, "max_tokens": max_tokens} + kwargs: dict[str, Any] = {"model": model} + # GPT-5.x / o-series reject legacy ``max_tokens``; they take + # ``max_completion_tokens``. Everything else keeps ``max_tokens``. + if _model_uses_max_completion_tokens(model): + kwargs["max_completion_tokens"] = max_tokens + else: + kwargs["max_tokens"] = max_tokens if _model_accepts_temperature(model): kwargs["temperature"] = temperature diff --git a/tests/unit/test_llm.py b/tests/unit/test_llm.py index 737b321a..27843a6d 100644 --- a/tests/unit/test_llm.py +++ b/tests/unit/test_llm.py @@ -4,7 +4,11 @@ """Unit tests for the custom OpenAI-compatible endpoint alternative (build_llm).""" from ad_buyer.config.settings import Settings -from ad_buyer.llm import _model_accepts_temperature, build_llm +from ad_buyer.llm import ( + _model_accepts_temperature, + _model_uses_max_completion_tokens, + build_llm, +) def _settings(**overrides) -> Settings: @@ -212,3 +216,59 @@ def test_openai_compatible_path_also_omits_temperature(self, monkeypatch): llm = build_llm(model="claude-opus-4-8", temperature=0.3, max_tokens=4096) assert llm.provider == "openai" assert llm.temperature is None + + +class TestModelUsesMaxCompletionTokens: + """OpenAI's GPT-5.x and o-series reasoning models reject the legacy + ``max_tokens`` parameter (400 invalid_request_error: 'max_tokens' is not + supported with this model) and take ``max_completion_tokens`` instead. + Other models keep ``max_tokens``.""" + + def test_gpt_5_uses_max_completion_tokens(self): + assert _model_uses_max_completion_tokens("us.openai.gpt-5.6-sol") is True + + def test_bare_gpt_5_uses_max_completion_tokens(self): + assert _model_uses_max_completion_tokens("openai/gpt-5") is True + + def test_o1_uses_max_completion_tokens(self): + assert _model_uses_max_completion_tokens("o1") is True + + def test_o3_uses_max_completion_tokens(self): + assert _model_uses_max_completion_tokens("openai/o3-mini") is True + + def test_match_is_case_insensitive(self): + assert _model_uses_max_completion_tokens("US.OpenAI.GPT-5.6-SOL") is True + + def test_gpt_4o_uses_legacy_max_tokens(self): + assert _model_uses_max_completion_tokens("openai/gpt-4o") is False + + def test_claude_uses_legacy_max_tokens(self): + assert _model_uses_max_completion_tokens("us.anthropic.claude-sonnet-5") is False + + +class TestMaxTokensRoutedToCorrectParam: + """build_llm routes the token cap to max_completion_tokens for models that + require it, and to max_tokens for everything else.""" + + def test_gpt_5_routes_to_max_completion_tokens(self, monkeypatch): + monkeypatch.setattr( + "ad_buyer.llm.get_settings", + lambda: _settings( + openai_compatible_llm_api_base_url="https://bedrock-runtime.us-west-2.amazonaws.com/openai/v1", + openai_compatible_llm_api_key="bedrock-key", + ), + ) + llm = build_llm(model="us.openai.gpt-5.6-sol", temperature=0.5, max_tokens=4096) + assert llm.max_completion_tokens == 4096 + assert llm.max_tokens is None + + def test_gpt_4o_keeps_max_tokens(self, monkeypatch): + monkeypatch.setattr( + "ad_buyer.llm.get_settings", + lambda: _settings( + openai_compatible_llm_api_base_url="https://bedrock-runtime.us-west-2.amazonaws.com/openai/v1", + openai_compatible_llm_api_key="bedrock-key", + ), + ) + llm = build_llm(model="gpt-4o", temperature=0.5, max_tokens=4096) + assert llm.max_tokens == 4096