diff --git a/python/legacy/src/x402/evm_paywall_template.py b/python/legacy/src/x402/evm_paywall_template.py
index 77b9de1ce..0e605c03e 100644
--- a/python/legacy/src/x402/evm_paywall_template.py
+++ b/python/legacy/src/x402/evm_paywall_template.py
@@ -1,2 +1,2 @@
# THIS FILE IS AUTO-GENERATED - DO NOT EDIT
-EVM_PAYWALL_TEMPLATE = "\n \n \n Payment Required\n \n \n \n \n "
+EVM_PAYWALL_TEMPLATE = '\n \n \n Payment Required\n \n \n \n \n '
diff --git a/python/legacy/src/x402/facilitator.py b/python/legacy/src/x402/facilitator.py
index 56c8ac1e5..981a20f79 100644
--- a/python/legacy/src/x402/facilitator.py
+++ b/python/legacy/src/x402/facilitator.py
@@ -1,3 +1,4 @@
+import inspect
from typing import Callable, Optional
from typing_extensions import (
TypedDict,
@@ -39,14 +40,23 @@ def __init__(self, config: Optional[FacilitatorConfig] = None):
self.config = {"url": url, "create_headers": config.get("create_headers")}
+ async def _get_custom_headers(self) -> dict[str, dict[str, str]] | None:
+ """Get custom headers, supporting both sync and async create_headers functions."""
+ if not self.config.get("create_headers"):
+ return None
+ result = self.config["create_headers"]()
+ if inspect.iscoroutine(result):
+ return await result
+ return result
+
async def verify(
self, payment: PaymentPayload, payment_requirements: PaymentRequirements
) -> VerifyResponse:
"""Verify a payment header is valid and a request should be processed"""
headers = {"Content-Type": "application/json"}
- if self.config.get("create_headers"):
- custom_headers = await self.config["create_headers"]()
+ custom_headers = await self._get_custom_headers()
+ if custom_headers:
headers.update(custom_headers.get("verify", {}))
async with httpx.AsyncClient() as client:
@@ -71,8 +81,8 @@ async def settle(
) -> SettleResponse:
headers = {"Content-Type": "application/json"}
- if self.config.get("create_headers"):
- custom_headers = await self.config["create_headers"]()
+ custom_headers = await self._get_custom_headers()
+ if custom_headers:
headers.update(custom_headers.get("settle", {}))
async with httpx.AsyncClient() as client:
@@ -107,8 +117,8 @@ async def list(
headers = {"Content-Type": "application/json"}
- if self.config.get("create_headers"):
- custom_headers = await self.config["create_headers"]()
+ custom_headers = await self._get_custom_headers()
+ if custom_headers:
headers.update(custom_headers.get("list", {}))
# Build query parameters, excluding None values
diff --git a/python/legacy/src/x402/svm_paywall_template.py b/python/legacy/src/x402/svm_paywall_template.py
index 766f149b1..7696ee2c0 100644
--- a/python/legacy/src/x402/svm_paywall_template.py
+++ b/python/legacy/src/x402/svm_paywall_template.py
@@ -1,2 +1,2 @@
# THIS FILE IS AUTO-GENERATED - DO NOT EDIT
-SVM_PAYWALL_TEMPLATE = "\n \n \n Payment Required\n \n \n \n \n "
+SVM_PAYWALL_TEMPLATE = '\n \n \n Payment Required\n \n \n \n \n '
diff --git a/python/x402/http/__init__.py b/python/x402/http/__init__.py
index c5db60433..221ef2b78 100644
--- a/python/x402/http/__init__.py
+++ b/python/x402/http/__init__.py
@@ -16,6 +16,7 @@
from .facilitator_client import (
AuthHeaders,
AuthProvider,
+ CreateHeadersAuthProvider,
FacilitatorClient,
FacilitatorConfig,
HTTPFacilitatorClient,
@@ -84,6 +85,7 @@
"FacilitatorConfig",
"AuthProvider",
"AuthHeaders",
+ "CreateHeadersAuthProvider",
# HTTP client
"x402HTTPClient",
"PaymentRoundTripper",
diff --git a/python/x402/http/facilitator_client.py b/python/x402/http/facilitator_client.py
index ac45824cc..74ab0b6e6 100644
--- a/python/x402/http/facilitator_client.py
+++ b/python/x402/http/facilitator_client.py
@@ -3,6 +3,7 @@
from __future__ import annotations
import json
+from collections.abc import Callable
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Any, Protocol
@@ -42,6 +43,26 @@ def get_auth_headers(self) -> AuthHeaders:
...
+class CreateHeadersAuthProvider:
+ """AuthProvider that wraps a create_headers callable.
+
+ Adapts the dict-style create_headers function (as used by CDP SDK)
+ to the AuthProvider protocol.
+ """
+
+ def __init__(self, create_headers: Callable[[], dict[str, dict[str, str]]]) -> None:
+ self._create_headers = create_headers
+
+ def get_auth_headers(self) -> AuthHeaders:
+ """Get authentication headers by calling the create_headers function."""
+ result = self._create_headers()
+ return AuthHeaders(
+ verify=result.get("verify", {}),
+ settle=result.get("settle", {}),
+ supported=result.get("supported", result.get("list", {})),
+ )
+
+
# ============================================================================
# FacilitatorClient Protocol
# ============================================================================
@@ -103,20 +124,37 @@ class HTTPFacilitatorClient:
Supports both V1 and V2 protocol versions.
"""
- def __init__(self, config: FacilitatorConfig | None = None) -> None:
+ def __init__(self, config: FacilitatorConfig | dict[str, Any] | None = None) -> None:
"""Create HTTP facilitator client.
Args:
- config: Optional configuration (uses defaults if not provided).
+ config: Optional configuration. Accepts either:
+ - FacilitatorConfig dataclass (recommended)
+ - Dict with 'url' and optional 'create_headers'
+ - None (uses defaults)
"""
- config = config or FacilitatorConfig()
-
- self._url = config.url.rstrip("/")
- self._timeout = config.timeout
- self._auth_provider = config.auth_provider
- self._identifier = config.identifier or self._url
- self._http_client = config.http_client
- self._owns_client = config.http_client is None
+ # Handle dict-style config
+ if isinstance(config, dict):
+ url = config.get("url", DEFAULT_FACILITATOR_URL)
+ create_headers = config.get("create_headers")
+ auth_provider = CreateHeadersAuthProvider(create_headers) if create_headers else None
+
+ self._url = url.rstrip("/")
+ self._timeout = 30.0
+ self._auth_provider = auth_provider
+ self._identifier = self._url
+ self._http_client = None
+ self._owns_client = True
+ else:
+ # Handle FacilitatorConfig dataclass or None
+ config = config or FacilitatorConfig()
+
+ self._url = config.url.rstrip("/")
+ self._timeout = config.timeout
+ self._auth_provider = config.auth_provider
+ self._identifier = config.identifier or self._url
+ self._http_client = config.http_client
+ self._owns_client = config.http_client is None
def _get_client(self) -> httpx.Client:
"""Get or create HTTP client."""
diff --git a/python/x402/http/x402_http_server.py b/python/x402/http/x402_http_server.py
index b21873b47..4fba41082 100644
--- a/python/x402/http/x402_http_server.py
+++ b/python/x402/http/x402_http_server.py
@@ -1063,5 +1063,3 @@ def _resolve_value_sync(
)
return result
return value
-
-
diff --git a/python/x402/tests/integrations/test_async_hooks.py b/python/x402/tests/integrations/test_async_hooks.py
index a85aa9208..1da3f1e7e 100644
--- a/python/x402/tests/integrations/test_async_hooks.py
+++ b/python/x402/tests/integrations/test_async_hooks.py
@@ -51,9 +51,7 @@ def get_body(self):
class TestAsyncHooks:
def setup_method(self):
- self.facilitator = x402Facilitator().register(
- ["x402:cash"], CashSchemeNetworkFacilitator()
- )
+ self.facilitator = x402Facilitator().register(["x402:cash"], CashSchemeNetworkFacilitator())
facilitator_client = CashFacilitatorClient(self.facilitator)
self.resource_server = x402ResourceServer(facilitator_client)
self.resource_server.register("x402:cash", CashSchemeNetworkServer())
diff --git a/python/x402/tests/integrations/test_core.py b/python/x402/tests/integrations/test_core.py
index f836aeb7a..dfae3f841 100644
--- a/python/x402/tests/integrations/test_core.py
+++ b/python/x402/tests/integrations/test_core.py
@@ -430,9 +430,7 @@ def test_middleware_verify_and_settle_cash_payment(self) -> None:
result.response.body,
)
payment_payload = self.http_client.create_payment_payload(payment_required)
- request_headers = self.http_client.encode_payment_signature_header(
- payment_payload
- )
+ request_headers = self.http_client.encode_payment_signature_header(payment_payload)
# Retry with payment
mock_adapter_with_payment = MockHTTPAdapter(
@@ -446,9 +444,7 @@ def test_middleware_verify_and_settle_cash_payment(self) -> None:
method="GET",
)
- result2 = asyncio.run(
- self.http_server.process_http_request(context_with_payment)
- )
+ result2 = asyncio.run(self.http_server.process_http_request(context_with_payment))
assert result2.type == "payment-verified"
assert result2.payment_payload is not None
assert result2.payment_requirements is not None