feat(python): handle McpError(-32042) for MCP payment challenges

python/x402/mcp/client.py

@@ -33,6 +33,7 @@
 from .constants import MCP_PAYMENT_META_KEY, MCP_PAYMENT_RESPONSE_META_KEY
 from .utils import (
     convert_mcp_result,
+    extract_payment_required_from_error,
     extract_payment_required_from_result,
     extract_payment_response_from_meta,
 )
@@ -241,7 +242,14 @@ def call_tool(
         args = args or {}
         params = {"name": name, "arguments": args}
 
-        result = self._mcp_client.call_tool(params, **kwargs)
+        try:
+            result = self._mcp_client.call_tool(params, **kwargs)
+        except Exception as exc:
+            payment_required = extract_payment_required_from_error(exc)
+            if payment_required is None:
+                raise
+            return self._handle_payment_required(name, args, payment_required, **kwargs)
+
         mcp_result = convert_mcp_result(result)
 
         payment_required = extract_payment_required_from_result(mcp_result)
@@ -251,12 +259,26 @@ def call_tool(
         if not self._auto_payment:
             return self._build_result(mcp_result, payment_made=False)
 
+        return self._handle_payment_required(name, args, payment_required, **kwargs)
+
+    def _handle_payment_required(
+        self,
+        name: str,
+        args: dict[str, Any],
+        payment_required: PaymentRequired,
+        **kwargs: Any,
+    ) -> MCPToolCallResult:
+        """Handle a payment-required signal (from isError result or thrown exception)."""
         if self._on_payment_requested:
             approved = self._on_payment_requested(
                 type("Ctx", (), {"payment_required": payment_required})()
             )
             if not approved:
-                return self._build_result(mcp_result, payment_made=False)
+                return MCPToolCallResult(
+                    content=[],
+                    is_error=True,
+                    payment_made=False,
+                )
 
         payment_payload = self._payment_client.create_payment_payload(payment_required)
         payload_dict = payment_payload.model_dump(by_alias=True)

python/x402/mcp/client_async.py

@@ -18,6 +18,7 @@
 from .utils import (
     attach_payment_to_meta,
     convert_mcp_result,
+    extract_payment_required_from_error,
     extract_payment_required_from_result,
     extract_payment_response_from_meta,
     register_schemes,
@@ -161,7 +162,14 @@ async def call_tool(
         """
         # First attempt without payment
         call_params = {"name": name, "arguments": args}
-        result = await self._call_mcp_tool(call_params, **kwargs)
+        try:
+            result = await self._call_mcp_tool(call_params, **kwargs)
+        except Exception as exc:
+            # Check if the thrown exception carries payment data (e.g. McpError -32042)
+            payment_required = extract_payment_required_from_error(exc)
+            if payment_required is None:
+                raise
+            return await self._handle_payment_required(name, args, payment_required, **kwargs)
 
         # Check if this is a payment required response
         payment_required = extract_payment_required_from_result(result)
@@ -174,7 +182,19 @@ async def call_tool(
                 payment_made=False,
             )
 
-        # Payment required - run hooks first
+        return await self._handle_payment_required(name, args, payment_required, **kwargs)
+
+    async def _handle_payment_required(
+        self,
+        name: str,
+        args: dict[str, Any],
+        payment_required: PaymentRequired,
+        **kwargs: Any,
+    ) -> MCPToolCallResult:
+        """Handle a payment-required signal (from isError result or thrown exception).
+
+        Runs hooks, checks auto_payment, creates payment, and retries.
+        """
         payment_required_context = PaymentRequiredContext(
             tool_name=name,
             arguments=args,
@@ -291,7 +311,10 @@ async def get_tool_payment_requirements(
             PaymentRequired if found, None otherwise
         """
         call_params = {"name": name, "arguments": args}
-        result = await self._call_mcp_tool(call_params, **kwargs)
+        try:
+            result = await self._call_mcp_tool(call_params, **kwargs)
+        except Exception as exc:
+            return extract_payment_required_from_error(exc)
         return extract_payment_required_from_result(result)
 
     async def _call_mcp_tool(self, params: dict[str, Any], **kwargs: Any) -> MCPToolResult: