feat(payments): Add LangGraph integration for payment handling (#546)

* feat(payments): Add LangGraph integration for payment handling

* refactor(payments): Unify LangGraph and Strands config into single class

  Merge AgentCorePaymentsConfig (LangGraph) and AgentCorePaymentsPluginConfig
  (Strands) into a single dataclass in integrations/config.py. Both names
  remain available as aliases for backward compatibility.

* define _ERROR_MESSAGES dict at the root level of payments SDK instead of in langgraph-specific package

* added langgraph dependencies to pyproject.toml for CI

* fix: linter check failing due to long lines, wrapped and fixed length of all

* style: apply ruff formatting locally for push and rerun lint check

* fix(payments): Fix broken type import and raise langchain/langgraph version floor

Fix TYPE_CHECKING import in errors.py to use ..config (parent package)
instead of .config (non-existent sibling module). This resolves mypy/pyright
failures for PaymentErrorContext.config type resolution.

Raise langchain and langgraph minimum versions from >=0.2.0 to >=1.0.0 in
both dev dependencies and the [langgraph] optional group.
AgentMiddleware, create_agent, and the langchain.agents.middleware namespace
are langchain 1.0 APIs — the 0.2.0 floor allowed installations that would
fail at import time.

* test(payments): Add unit tests for auto_session lazy session creation

Cover the auto_session feature path that was previously only validated
live against testnet. Tests verify:
- Session created on first 402 when auto_session=True
- Config mutated with new session ID for subsequent calls
- Session reused across multiple tool calls (no duplicate creation)
- auto_session=False still raises PAYMENT ERROR without session
- Budget and expiry config values passed correctly
- Pre-existing session_id skips auto-creation

* docs(payments): Document pre-merge testnet validation requirement

Add PRE-MERGE REQUIREMENT note to test_functional.py docstring clarifying
that any middleware changes must be validated against live testnet before
merge, since these tests are skipped in CI.

* docs(payments): Document middleware instance lifecycle and thread-safety

Add note that one middleware instance should be created per agent
invocation/request. The middleware is not thread-safe due to config
mutations in auto_session and on_payment_error callbacks.

* refactor(payments): Extract shared helpers to reduce sync/async duplication

Extract guard checks, 402 detection, payment request extraction, header
injection, and post-payment rejection detection into shared private
methods. The sync and async paths now only differ at await/sleep
boundaries. Future bug fixes to detection or injection logic only need
to be applied in one place instead of four.

No behavioral changes — 128 tests pass identically before and after.

* docs(payments): Add Sync vs Async section to langgraph middleware README

Document when each path is used (.invoke vs .ainvoke), what the async
path does differently (non-blocking sleep, to_thread for signing, async
callbacks), and provide FastAPI and script examples.

* fix(payments): Detect async callback on sync path and fail loudly

If an async def on_payment_error callback is registered but the agent
runs via .invoke() (sync path), the callback would silently return an
unawaited coroutine, bypass RETRY logic, and leak a RuntimeWarning.

Now we detect this with inspect.iscoroutinefunction() and raise a clear
TypeError that the existing try/except catches and logs — the agent
continues with the default error message and the developer sees exactly
what to fix in their logs.

* test(payments): Cover post-recovery rejection with raw JSON fallback path

Verify that when a raw-JSON tool returns 402 after error handler
recovery, the FallbackHandler is used to extract the real error detail
(e.g. 'budget exceeded') instead of falling through to 'unknown'.

* fix(payments): Reassign handler in post-recovery fallback detection

When fallback detects a 402 in the recovery retry path, reassign _rh to
_FallbackHandler(fallback) so extract_body returns the actual parsed
body. Without this, GenericPaymentHandler looks for the PAYMENT_REQUIRED
marker, finds nothing in raw JSON, and the LLM sees 'unknown' instead
of the real error detail.

* fix(payments): Add name-based handler as third 402 detection fallback

When both GenericPaymentHandler (marker) and _fallback_detect_402 (JSON)
fail to detect a 402, try the name-based handler from get_payment_handler
(e.g. HttpRequestPaymentHandler for tools named http_request). This
covers the legacy 'Status Code: 402' text-block format used by Strands
tools ported to LangGraph without format adaptation.

* fix(payments): Feed custom handlers raw content instead of prepared shape

Custom handlers now receive result.content (the raw ToolMessage content)
for extract_status_code, extract_headers, and extract_body — not the
internal {'content': [{'text': ...}]} prepared shape.

This makes the custom handler contract intuitive: handlers parse the
tool's actual output format, not a middleware-internal wrapper. Fixes
silent detection failures when custom handlers expected raw JSON or
other native formats.

* test(payments): Fill async path and custom handler test coverage gaps

Add tests for:
- Async auto_session creation and reuse
- Async post-payment rejection with raw JSON fallback
- Async name-based handler fallback (legacy text-block format)
- Async custom handler receiving raw content
- Async error handler callback (retry and propagate paths)
- Sync custom handler raw content contract verification

Suite now at 141 passed, providing async path parity with sync tests.

* refactor(payments): Collapse remaining sync/async duplication

- Merge _check_post_payment_rejection and _check_post_recovery_rejection
  into single _check_retry_rejection with context parameter
- Extract _inject_for_error_retry for shared injection in error handlers
- Extract _build_error_context for shared PaymentErrorContext construction
- Extract _handle_callback_resolution for shared resolution dispatch

Reduces middleware.py from 805 to 704 lines. The sync/async error
handlers now only differ at await/sleep/to_thread boundaries.
141 tests pass identically.

* refactor(payments): Hoist all inline imports to module top

Move asyncio, json, GenericPaymentHandler, ErrorResolution, and
PaymentErrorContext imports to the top of the file. None have circular
dependency risks. Eliminates repeated imports that contributed to
drift between duplicated code blocks.

* fix(payments): Update functional test custom handlers for raw content contract

Update TrackingHandler and RawJsonHandler in test_functional.py to
handle raw content (str/list) instead of the prepared shape dict.
Required after the custom handler contract change in 076d6ed.

---------

Co-authored-by: Aidan Daly <99039782+aidandaly24@users.noreply.github.com>

Author: ragsu43

pyproject.toml

@@ -151,6 +151,9 @@ dev = [
     "wheel>=0.45.1",
     "strands-agents>=1.20.0",
     "strands-agents-evals>=0.1.0",
+    "langchain>=1.0.0",
+    "langgraph>=1.0.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",
@@ -163,6 +166,12 @@ strands-agents = [
     "strands-agents>=1.20.0",
     "mcp>=1.23.0,<2.0.0",
 ]
+langgraph = [
+    "langchain>=1.0.0",
+    "langgraph>=1.0.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."
+    )