feat(payments): Add LangGraph integration for payment handling

pyproject.toml

@@ -154,6 +154,9 @@ dev = [
     "wheel>=0.45.1",
     "strands-agents>=1.20.0",
     "strands-agents-evals>=0.1.0",
+    "langchain>=0.2.0",
+    "langgraph>=0.2.0",
+    "langchain-mcp-adapters>=0.1.0",
     "a2a-sdk[http-server]>=0.3,<1.0",
     "ag-ui-protocol>=0.1.10",
     "mcp-proxy-for-aws>=0.1.0",
@@ -166,6 +169,12 @@ strands-agents = [
     "strands-agents>=1.20.0",
     "mcp>=1.23.0,<2.0.0",
 ]
+langgraph = [
+    "langchain>=0.2.0",
+    "langgraph>=0.2.0",
+    "langchain-mcp-adapters>=0.1.0",
+    "httpx>=0.27.0",
+]
 strands-agents-evals = [
     "strands-agents-evals>=0.1.0"
 ]

src/bedrock_agentcore/payments/integrations/config.py

@@ -1,62 +1,49 @@
-"""Configuration for AgentCorePaymentsPlugin."""
+"""Configuration for AgentCore Payments integrations (Strands and LangGraph)."""
 
-from dataclasses import dataclass
-from typing import Callable, List, Optional
+from dataclasses import dataclass, field
+from typing import Any, Callable, Dict, List, Optional
+
+from .handlers import PaymentResponseHandler
 
 
 @dataclass
 class AgentCorePaymentsPluginConfig:
-    """Configuration for AgentCorePaymentsPlugin.
+    """Configuration for AgentCore Payments integrations.
+
+    This unified config is used by both the Strands plugin and LangGraph middleware.
 
     Attributes:
-        payment_manager_arn: ARN of the payment manager service
-        region: AWS region for the payment manager
+        payment_manager_arn: ARN of the payment manager service.
         user_id: User ID for payment processing. Required for SigV4 auth.
             Optional for bearer token auth (JWT identifies the user).
             When set with bearer auth, propagated via X-Amzn-Bedrock-AgentCore-Payments-User-Id header.
         payment_instrument_id: Optional payment instrument ID for the user.
             Can be set later via update_payment_instrument_id().
         payment_session_id: Optional payment session ID for the transaction.
             Can be set later via update_payment_session_id().
-        network_preferences_config: Optional list of network CAIP2 identifiers
-            in order of preference. If not provided, defaults to the system default.
-        auto_payment: Whether to automatically process 402 payment requirements.
-            Defaults to True to maintain existing behavior.
-        max_interrupt_retries: Maximum number of interrupt retries per tool use.
-            Defaults to 5. Set to 0 to disable interrupt retries entirely (no interrupts will be raised).
-        agent_name: Optional agent name to propagate via the
-            X-Amzn-Bedrock-AgentCore-Payments-Agent-Name HTTP header on every
-            AgentCore payments data-plane API call. When set, the header is automatically injected
-            by PaymentManager and propagated for Payments.
-        bearer_token: Optional static JWT bearer token for OAuth/CUSTOM_JWT authentication.
-            When set, PaymentManager uses Bearer token auth instead of SigV4.
-            Mutually exclusive with token_provider.
-        token_provider: Optional callable that returns a fresh JWT bearer token string.
-            Called before each request to support token refresh.
-            Mutually exclusive with bearer_token.
-        payment_tool_allowlist: Optional list of tool names that are eligible for
-            automatic X402 payment processing. When None (default), all tools are
-            eligible (preserving existing behavior). When set, only tool calls whose
-            name appears in this list will trigger payment processing; all others are
-            skipped.
-        provide_http_request: Whether the plugin should register its built-in
-            ``http_request`` ``@tool`` on the agent. Defaults to True so adding the
-            plugin gives a turnkey paid-HTTP experience. Set to False if you want
-            to ship your own ``http_request`` tool — Strands raises a ValueError
-            on duplicate tool names, so you must opt out of the plugin's version
-            before passing your own. Auto-payment of 402 responses still works
-            against any tool whose output carries the ``PAYMENT_REQUIRED:``
-            content marker, so disabling this flag does not disable interception.
-        post_payment_retry_delay_seconds: Seconds to wait after generating a
-            payment header before allowing the tool to be retried. The x402
-            EIP-3009 ``transferWithAuthorization`` contract requires
-            ``block.timestamp > validAfter`` (strict greater-than). Some signing
-            services set ``validAfter`` close to the current time, which can
-            cause the merchant facilitator to submit before ``validAfter``
-            elapses, producing a misleading "invalid_payload" response. A small
-            delay between signing and retry lets the chain advance one block so
-            the authorization is valid by the time the seller submits. Defaults
-            to 3.0 seconds (about one Base Sepolia block). Set to 0 to disable.
+        payment_connector_id: Payment connector ID (optional).
+        region: AWS region for the payment manager.
+        network_preferences_config: Ordered list of network CAIP2 identifiers.
+        auto_payment: Whether to automatically process 402 responses. Default True.
+        agent_name: Agent name propagated via HTTP header on data-plane calls.
+        bearer_token: Static JWT for OAuth/CUSTOM_JWT auth. Mutually exclusive with token_provider.
+        token_provider: Callable returning a fresh JWT. Mutually exclusive with bearer_token.
+        payment_tool_allowlist: Tool names eligible for payment processing. None = all tools.
+        provide_http_request: Whether the integration registers its built-in http_request tool.
+        post_payment_retry_delay_seconds: Delay after signing before retry. Default 3.0s.
+        max_interrupt_retries: Maximum number of interrupt retries per tool use (Strands only).
+            Defaults to 5. Set to 0 to disable interrupt retries entirely.
+        custom_handlers: Custom PaymentResponseHandler instances keyed by tool name.
+            Takes precedence over the built-in handler registry during resolution.
+        auto_session: Whether to auto-create a payment session on first 402 if
+            payment_session_id is not set. Default False.
+        auto_session_budget: Budget for auto-created sessions (USD). Default "1.00".
+        auto_session_expiry_minutes: Expiry time for auto-created sessions. Default 60.
+        on_payment_error: Optional callback invoked when a payment exception occurs.
+            Receives PaymentErrorContext, returns ErrorResolution.RETRY or .PROPAGATE.
+            When None (default), errors produce deterministic ToolMessages directly.
+        max_error_retries: Maximum times the error callback can return RETRY per tool call.
+            Default 3. Set to 0 to disable the callback entirely.
     """
 
     payment_manager_arn: str
@@ -65,51 +52,52 @@ class AgentCorePaymentsPluginConfig:
     payment_session_id: Optional[str] = None
     payment_connector_id: Optional[str] = None
     region: Optional[str] = None
-    network_preferences_config: Optional[list[str]] = None
+    network_preferences_config: Optional[List[str]] = None
     auto_payment: bool = True
-    max_interrupt_retries: int = 5
     agent_name: Optional[str] = None
     bearer_token: Optional[str] = None
     token_provider: Optional[Callable[[], str]] = None
     payment_tool_allowlist: Optional[List[str]] = None
     provide_http_request: bool = True
     post_payment_retry_delay_seconds: float = 3.0
+    max_interrupt_retries: int = 5
+    custom_handlers: Optional[Dict[str, Any]] = field(default=None)
+    auto_session: bool = False
+    auto_session_budget: str = "1.00"
+    auto_session_expiry_minutes: int = 60
+    on_payment_error: Optional[Callable] = None
+    max_error_retries: int = 3
 
     def __post_init__(self) -> None:
         """Validate configuration after initialization."""
         if not self.payment_manager_arn:
             raise ValueError("payment_manager_arn is required")
-
         if not self.payment_manager_arn.startswith("arn:"):
             raise ValueError(f"Invalid ARN format: {self.payment_manager_arn}")
 
+        if self.bearer_token is not None and self.token_provider is not None:
+            raise ValueError("bearer_token and token_provider are mutually exclusive")
         if self.bearer_token is not None and not isinstance(self.bearer_token, str):
             raise ValueError(f"bearer_token must be a string, got {type(self.bearer_token).__name__}")
-
         if self.token_provider is not None and not callable(self.token_provider):
             raise ValueError(f"token_provider must be callable, got {type(self.token_provider).__name__}")
 
-        if self.user_id is not None and self.user_id and not self.user_id.strip():
-            raise ValueError("user_id cannot be whitespace-only")
-
         if not self.user_id and self.bearer_token is None and self.token_provider is None:
             raise ValueError("user_id is required for SigV4 auth (when bearer_token/token_provider not set)")
+        if self.user_id is not None and self.user_id and not self.user_id.strip():
+            raise ValueError("user_id cannot be whitespace-only")
 
         if not isinstance(self.auto_payment, bool):
             raise ValueError(f"auto_payment must be a boolean, got {type(self.auto_payment).__name__}")
-
-        if self.bearer_token is not None and self.token_provider is not None:
-            raise ValueError("bearer_token and token_provider are mutually exclusive. Provide only one.")
+        if not isinstance(self.provide_http_request, bool):
+            raise ValueError(f"provide_http_request must be a boolean, got {type(self.provide_http_request).__name__}")
 
         if self.payment_tool_allowlist is not None:
             if not isinstance(self.payment_tool_allowlist, list):
                 raise ValueError("payment_tool_allowlist must be a list of tool name strings")
             if not all(isinstance(t, str) for t in self.payment_tool_allowlist):
                 raise ValueError("All entries in payment_tool_allowlist must be strings")
 
-        if not isinstance(self.provide_http_request, bool):
-            raise ValueError(f"provide_http_request must be a boolean, got {type(self.provide_http_request).__name__}")
-
         if not isinstance(self.post_payment_retry_delay_seconds, (int, float)) or isinstance(
             self.post_payment_retry_delay_seconds, bool
         ):
@@ -122,6 +110,24 @@ def __post_init__(self) -> None:
                 f"post_payment_retry_delay_seconds must be >= 0, got {self.post_payment_retry_delay_seconds}"
             )
 
+        if self.custom_handlers is not None:
+            if not isinstance(self.custom_handlers, dict):
+                raise ValueError(
+                    "custom_handlers must be a dict mapping tool names to PaymentResponseHandler instances"
+                )
+            if not all(isinstance(k, str) for k in self.custom_handlers):
+                raise ValueError("All keys in custom_handlers must be strings")
+            if not all(isinstance(v, PaymentResponseHandler) for v in self.custom_handlers.values()):
+                raise ValueError("All values in custom_handlers must be PaymentResponseHandler instances")
+
+        if self.on_payment_error is not None and not callable(self.on_payment_error):
+            raise ValueError(f"on_payment_error must be callable, got {type(self.on_payment_error).__name__}")
+
+        if not isinstance(self.max_error_retries, int) or isinstance(self.max_error_retries, bool):
+            raise ValueError(f"max_error_retries must be an int, got {type(self.max_error_retries).__name__}")
+        if self.max_error_retries < 0:
+            raise ValueError(f"max_error_retries must be >= 0, got {self.max_error_retries}")
+
     def update_payment_session_id(self, payment_session_id: str) -> None:
         """Update the payment session ID.
 
@@ -141,3 +147,40 @@ def update_payment_instrument_id(self, payment_instrument_id: str) -> None:
         if not payment_instrument_id:
             raise ValueError("payment_instrument_id cannot be empty")
         self.payment_instrument_id = payment_instrument_id
+
+    def add_to_allowlist(self, *tool_names: str) -> None:
+        """Add tool names to the payment allowlist.
+
+        Creates the allowlist if it doesn't exist yet (switching from "all tools"
+        to explicit allowlist mode).
+
+        Args:
+            tool_names: One or more tool names to add.
+        """
+        if self.payment_tool_allowlist is None:
+            self.payment_tool_allowlist = []
+        for name in tool_names:
+            if not isinstance(name, str):
+                raise ValueError(f"Tool name must be a string, got {type(name).__name__}")
+            if name not in self.payment_tool_allowlist:
+                self.payment_tool_allowlist.append(name)
+
+    def remove_from_allowlist(self, *tool_names: str) -> None:
+        """Remove tool names from the payment allowlist.
+
+        If the allowlist becomes empty, sets it to None (all tools eligible).
+
+        Args:
+            tool_names: One or more tool names to remove.
+        """
+        if self.payment_tool_allowlist is None:
+            return
+        for name in tool_names:
+            if name in self.payment_tool_allowlist:
+                self.payment_tool_allowlist.remove(name)
+        if not self.payment_tool_allowlist:
+            self.payment_tool_allowlist = None
+
+
+# Backward-compatible alias for LangGraph imports
+AgentCorePaymentsConfig = AgentCorePaymentsPluginConfig

src/bedrock_agentcore/payments/integrations/error_messages.py

@@ -0,0 +1,74 @@
+"""Shared deterministic error messages for payment exceptions.
+
+These messages are designed to be shown to LLMs via tool results. They instruct
+the model not to retry and to inform the user of the specific issue.
+
+Used by: LangGraph middleware, and available for any future plugin integration.
+"""
+
+from typing import Dict, Type
+
+from bedrock_agentcore.payments.manager import (
+    InsufficientBudget,
+    PaymentError,
+    PaymentInstrumentConfigurationRequired,
+    PaymentInstrumentNotFound,
+    PaymentSessionConfigurationRequired,
+    PaymentSessionExpired,
+    PaymentSessionNotFound,
+)
+
+# Maps exception types to deterministic, LLM-instructive messages.
+PAYMENT_ERROR_MESSAGES: Dict[Type[Exception], str] = {
+    PaymentInstrumentConfigurationRequired: (
+        "No payment instrument configured. Do not retry this call. "
+        "Inform the user they need to configure a payment instrument before making paid requests."
+    ),
+    PaymentSessionConfigurationRequired: (
+        "No payment session configured. Do not retry this call. "
+        "Inform the user they need to create a payment session before making paid requests."
+    ),
+    PaymentInstrumentNotFound: (
+        "Payment instrument not found. Do not retry this call. "
+        "Inform the user their payment instrument ID is invalid or has been deleted."
+    ),
+    PaymentSessionNotFound: (
+        "Payment session not found. Do not retry this call. "
+        "Inform the user their payment session ID is invalid or has expired."
+    ),
+    PaymentSessionExpired: (
+        "Payment session has expired. Do not retry this call. "
+        "Inform the user they need to create a new payment session."
+    ),
+    InsufficientBudget: (
+        "Insufficient budget. The payment amount exceeds the remaining session limit. "
+        "Do not retry this call. Inform the user they need to increase their session budget "
+        "or create a new session with higher limits."
+    ),
+}
+
+
+def get_payment_error_message(exception: Exception) -> str:
+    """Get the deterministic error message for a payment exception.
+
+    Looks up the exception type in the message map. Falls back to a generic
+    message that includes the exception string for unrecognized types.
+
+    Args:
+        exception: The payment exception.
+
+    Returns:
+        Human/LLM-readable error message string.
+    """
+    msg = PAYMENT_ERROR_MESSAGES.get(type(exception))
+    if msg is not None:
+        return msg
+    if isinstance(exception, PaymentError):
+        return (
+            f"Payment processing failed ({exception}). "
+            "Do not retry this call. Inform the user that payment could not be completed."
+        )
+    return (
+        f"An unexpected error occurred during payment processing ({exception}). "
+        "Do not retry this call. Inform the user that payment could not be completed."
+    )