fix(python/mcp): gate tool output on settlement success

python/x402/mcp/server_async.py

@@ -272,6 +272,14 @@ async def wrapped_handler(args: dict[str, Any], extra: dict[str, Any]) -> MCPToo
                     resource_server, tool_name, config, str(e)
                 )
 
+            if not settle_result.success:
+                return await _create_settlement_failed_result_async(
+                    resource_server,
+                    tool_name,
+                    config,
+                    settle_result.error_reason or "Unknown settlement failure",
+                )
+
             # Run onAfterSettlement hook if present
             if config.hooks and config.hooks.on_after_settlement:
                 settlement_context = SettlementContext(

python/x402/mcp/server_sync.py

@@ -158,6 +158,14 @@ def wrapped_handler(args: dict[str, Any], extra: dict[str, Any]) -> MCPToolResul
                     resource_server, tool_name, config, str(e)
                 )
 
+            if not settle_result.success:
+                return _create_settlement_failed_result_sync(
+                    resource_server,
+                    tool_name,
+                    config,
+                    settle_result.error_reason or "Unknown settlement failure",
+                )
+
             if config.hooks and config.hooks.on_after_settlement:
                 settlement_context = SettlementContext(
                     tool_name=tool_name,

python/x402/mcp/tests/test_server.py

@@ -59,7 +59,9 @@ def find_matching_requirements(self, available, payload):
                 return req
         return None
 
-    def _create_payment_required_response_real(self, accepts, resource_info, error_msg):
+    def _create_payment_required_response_real(
+        self, accepts, resource_info, error_msg, extensions=None
+    ):
         """Real implementation of create payment required response."""
         from x402.schemas import PaymentRequired
 
@@ -352,6 +354,78 @@ def handler(args, context):
     assert "settlement" in str(result.content).lower() or result.structured_content is not None
 
 
+def test_create_payment_wrapper_settlement_returns_failure():
+    """Replay attack regression: settle returning success=False must withhold tool output."""
+    from x402.mcp.types import MCP_PAYMENT_RESPONSE_META_KEY
+
+    server = MockResourceServer()
+    server.settle_payment = Mock(
+        return_value=SettleResponse(
+            success=False,
+            error_reason="AuthorizationAlreadyUsed",
+            transaction="",
+            network="eip155:84532",
+        )
+    )
+
+    after_settlement_calls = []
+
+    def on_after_settlement(ctx):
+        after_settlement_calls.append(ctx)
+
+    config = PaymentWrapperConfig(
+        accepts=[
+            PaymentRequirements(
+                scheme="exact",
+                network="eip155:84532",
+                amount="1000",
+                asset="USDC",
+                pay_to="0xrecipient",
+                max_timeout_seconds=300,
+            )
+        ],
+        hooks=PaymentWrapperHooks(on_after_settlement=on_after_settlement),
+    )
+
+    paid = create_payment_wrapper(server, config)
+
+    def handler(args, context):
+        return {"content": [{"type": "text", "text": "should-not-be-returned"}]}
+
+    wrapped = paid(handler)
+    payload = PaymentPayload(
+        x402_version=2,
+        accepted={
+            "scheme": "exact",
+            "network": "eip155:84532",
+            "amount": "1000",
+            "asset": "USDC",
+            "pay_to": "0xrecipient",
+            "max_timeout_seconds": 300,
+        },
+        payload={"signature": "0x123"},
+    )
+    result = wrapped(
+        {"test": "value"},
+        {
+            "_meta": {
+                "x402/payment": (
+                    payload.model_dump() if hasattr(payload, "model_dump") else payload
+                )
+            }
+        },
+    )
+
+    assert result.is_error is True
+    assert "should-not-be-returned" not in str(result.content)
+    assert "should-not-be-returned" not in str(result.structured_content)
+    assert result.structured_content is not None
+    payment_response = result.structured_content[MCP_PAYMENT_RESPONSE_META_KEY]
+    assert payment_response["success"] is False
+    assert "AuthorizationAlreadyUsed" in result.structured_content["error"]
+    assert after_settlement_calls == []
+
+
 def test_create_payment_wrapper_handler_error_no_settlement():
     """Test that settlement is NOT called when handler returns an error."""
     server = MockResourceServer()

python/x402/mcp/tests/test_server_async.py

@@ -48,7 +48,9 @@ def find_matching_requirements(self, available, payload):
                 return req
         return None
 
-    async def _create_payment_required_response_real(self, accepts, resource_info, error_msg):
+    async def _create_payment_required_response_real(
+        self, accepts, resource_info, error_msg, extensions=None
+    ):
         """Real implementation of create payment required response."""
         from x402.schemas import PaymentRequired
 
@@ -379,6 +381,79 @@ async def handler(args, context):
     assert "settlement" in str(result.content).lower() or result.structured_content is not None
 
 
+@pytest.mark.asyncio
+async def test_create_payment_wrapper_async_settlement_returns_failure():
+    """Replay attack regression: settle returning success=False must withhold tool output."""
+    from x402.mcp.types import MCP_PAYMENT_RESPONSE_META_KEY
+
+    server = MockAsyncResourceServer()
+    server.settle_payment = AsyncMock(
+        return_value=SettleResponse(
+            success=False,
+            error_reason="AuthorizationAlreadyUsed",
+            transaction="",
+            network="eip155:84532",
+        )
+    )
+
+    after_settlement_calls = []
+
+    async def on_after_settlement(ctx):
+        after_settlement_calls.append(ctx)
+
+    config = PaymentWrapperConfig(
+        accepts=[
+            PaymentRequirements(
+                scheme="exact",
+                network="eip155:84532",
+                amount="1000",
+                asset="USDC",
+                pay_to="0xrecipient",
+                max_timeout_seconds=300,
+            )
+        ],
+        hooks=PaymentWrapperHooks(on_after_settlement=on_after_settlement),
+    )
+
+    paid = create_payment_wrapper(server, config)
+
+    async def handler(args, context):
+        return {"content": [{"type": "text", "text": "should-not-be-returned"}]}
+
+    wrapped = paid(handler)
+    payload = PaymentPayload(
+        x402_version=2,
+        accepted={
+            "scheme": "exact",
+            "network": "eip155:84532",
+            "amount": "1000",
+            "asset": "USDC",
+            "pay_to": "0xrecipient",
+            "max_timeout_seconds": 300,
+        },
+        payload={"signature": "0x123"},
+    )
+    result = await wrapped(
+        {"test": "value"},
+        {
+            "_meta": {
+                "x402/payment": (
+                    payload.model_dump() if hasattr(payload, "model_dump") else payload
+                )
+            }
+        },
+    )
+
+    assert result.is_error is True
+    assert "should-not-be-returned" not in str(result.content)
+    assert "should-not-be-returned" not in str(result.structured_content)
+    assert result.structured_content is not None
+    payment_response = result.structured_content[MCP_PAYMENT_RESPONSE_META_KEY]
+    assert payment_response["success"] is False
+    assert "AuthorizationAlreadyUsed" in result.structured_content["error"]
+    assert after_settlement_calls == []
+
+
 @pytest.mark.asyncio
 async def test_create_payment_wrapper_async_handler_error_no_settlement():
     """Test that settlement is NOT called when async handler returns an error."""