Skip to content

refactor: improve X402Auth streaming stability and schema filtering #147

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
lofingv wants to merge 6 commits into
base: main
Choose a base branch
from
Open

refactor: improve X402Auth streaming stability and schema filtering #147

Changes from 1 commit
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
6730a63
refactor: update X402Auth with better stream handling and schema filt…
lofingv Feb 5, 2026
0a5ad6a
refactor: improve type safety and use scheme_filter in auth flow
lofingv Feb 6, 2026
1394d14
chore: remove deprecated x402_auth.py after upstream refactor
lofingv Feb 7, 2026
08746a0
refactor: move x402_auth to client folder and add filters
lofingv Feb 7, 2026
bbf5b7d
refactor: move x402_auth to client folder and add filters
lofingv Feb 7, 2026
2f8b52a
Merge branch 'main' into audit/x402-auth-security
lofingv Feb 7, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 10 additions & 19 deletions src/opengradient/x402_auth.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,18 +31,9 @@ def __init__(
self,
account: typing.Any,
max_value: typing.Optional[int] = None,
payment_requirements_selector: typing.Optional[
typing.Callable[
[
list[PaymentRequirements],
typing.Optional[str],
typing.Optional[str],
typing.Optional[int],
],
PaymentRequirements,
]
] = None,
payment_requirements_selector: typing.Optional[typing.Callable] = None,
network_filter: typing.Optional[str] = None,
scheme_filter: typing.Optional[str] = None,
):
"""
Initialize X402Auth with an Ethereum account for signing payments.
Expand All @@ -56,22 +47,18 @@ def __init__(
self.x402_client = x402Client(
account,
max_value=max_value,
payment_requirements_selector=payment_requirements_selector, # type: ignore
payment_requirements_selector=payment_requirements_selector,
)
self.network_filter = network_filter
self.scheme_filter = scheme_filter

async def async_auth_flow(
self, request: httpx.Request
) -> typing.AsyncGenerator[httpx.Request, httpx.Response]:
"""
Handle authentication flow for x402 payment protocol.

Args:
request: httpx Request object to be authenticated

Yields:
httpx Request object with authentication headers attached
"""
request.read()
response = yield request

if response.status_code == 402:
Expand All @@ -85,6 +72,10 @@ async def async_auth_flow(
payment_response.accepts,
self.network_filter,
)

if not selected_requirements:
logging.error("X402Auth: No compatible payment requirements found")
return

payment_header = self.x402_client.create_payment_header(
selected_requirements, payment_response.x402_version
Expand All @@ -95,5 +86,5 @@ async def async_auth_flow(
yield request

except Exception as e:
logging.error(f"X402Auth: Error handling payment: {e}")
logging.error(f"X402Auth: Error handling payment: {str(e)}")
return