Skip to content

Gateways

BaseGateway

paygraph.gateways.base.BaseGateway

Bases: ABC

Abstract base class for all payment gateways.

Subclass this to implement a custom gateway. You must implement execute(). Override execute_async() for native async support and revoke() for card-style gateways that support cancellation.

Source code in src/paygraph/gateways/base.py 
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
class BaseGateway(ABC):
    """Abstract base class for all payment gateways.

    Subclass this to implement a custom gateway. You must implement
    ``execute()``. Override ``execute_async()`` for native async support
    and ``revoke()`` for card-style gateways that support cancellation.
    """

    @abstractmethod
    def execute(
        self, amount_cents: int, vendor: str, memo: str, **kwargs
    ) -> SpendResult:
        """Execute a spend for the given amount.

        Subclasses should declare gateway-specific parameters as explicit
        keyword-only arguments rather than consuming ``**kwargs``. This
        ensures callers get immediate feedback on typos.

        Args:
            amount_cents: Spend amount in cents.
            vendor: Name of the vendor.
            memo: Justification or memo for the spend.

        Returns:
            A ``SpendResult`` (or subclass) with the transaction details.
        """
        ...

    async def execute_async(
        self, amount_cents: int, vendor: str, memo: str, **kwargs
    ) -> SpendResult:
        """Execute a spend asynchronously.

        Default implementation runs ``execute()`` in a thread pool.
        Override for native async support (e.g. x402 gateways).

        Args:
            amount_cents: Spend amount in cents.
            vendor: Name of the vendor.
            memo: Justification or memo for the spend.

        Returns:
            A ``SpendResult`` (or subclass) with the transaction details.
        """
        loop = asyncio.get_running_loop()
        return await loop.run_in_executor(
            None, lambda: self.execute(amount_cents, vendor, memo, **kwargs)
        )

    def revoke(self, gateway_ref: str) -> bool:
        """Revoke (cancel) a previously issued spend.

        Optional — card gateways override this, x402 gateways typically don't.

        Args:
            gateway_ref: The ``gateway_ref`` from the ``SpendResult``.

        Returns:
            True if successfully revoked, False if not found.

        Raises:
            NotImplementedError: If this gateway does not support revocation.
        """
        raise NotImplementedError(
            f"{type(self).__name__} does not support revoke"
        )

execute(amount_cents, vendor, memo, **kwargs) abstractmethod

Execute a spend for the given amount.

Subclasses should declare gateway-specific parameters as explicit keyword-only arguments rather than consuming **kwargs. This ensures callers get immediate feedback on typos.

Parameters:

Name Type Description Default
amount_cents int

Spend amount in cents.

 required
vendor str

Name of the vendor.

 required
memo str

Justification or memo for the spend.

 required

Returns:

Type Description
SpendResult

A SpendResult (or subclass) with the transaction details.
Source code in src/paygraph/gateways/base.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
@abstractmethod
def execute(
    self, amount_cents: int, vendor: str, memo: str, **kwargs
) -> SpendResult:
    """Execute a spend for the given amount.

    Subclasses should declare gateway-specific parameters as explicit
    keyword-only arguments rather than consuming ``**kwargs``. This
    ensures callers get immediate feedback on typos.

    Args:
        amount_cents: Spend amount in cents.
        vendor: Name of the vendor.
        memo: Justification or memo for the spend.

    Returns:
        A ``SpendResult`` (or subclass) with the transaction details.
    """
    ...

execute_async(amount_cents, vendor, memo, **kwargs) async

Execute a spend asynchronously.

Default implementation runs execute() in a thread pool. Override for native async support (e.g. x402 gateways).

Parameters:

Name Type Description Default
amount_cents int

Spend amount in cents.

 required
vendor str

Name of the vendor.

 required
memo str

Justification or memo for the spend.

 required

Returns:

Type Description
SpendResult

A SpendResult (or subclass) with the transaction details.
Source code in src/paygraph/gateways/base.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
async def execute_async(
    self, amount_cents: int, vendor: str, memo: str, **kwargs
) -> SpendResult:
    """Execute a spend asynchronously.

    Default implementation runs ``execute()`` in a thread pool.
    Override for native async support (e.g. x402 gateways).

    Args:
        amount_cents: Spend amount in cents.
        vendor: Name of the vendor.
        memo: Justification or memo for the spend.

    Returns:
        A ``SpendResult`` (or subclass) with the transaction details.
    """
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(
        None, lambda: self.execute(amount_cents, vendor, memo, **kwargs)
    )

revoke(gateway_ref)

Revoke (cancel) a previously issued spend.

Optional — card gateways override this, x402 gateways typically don't.

Parameters:

Name Type Description Default
gateway_ref str

The gateway_ref from the SpendResult.

 required

Returns:

Type Description
bool

True if successfully revoked, False if not found.

Raises:

Type Description
NotImplementedError

If this gateway does not support revocation.
Source code in src/paygraph/gateways/base.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
def revoke(self, gateway_ref: str) -> bool:
    """Revoke (cancel) a previously issued spend.

    Optional — card gateways override this, x402 gateways typically don't.

    Args:
        gateway_ref: The ``gateway_ref`` from the ``SpendResult``.

    Returns:
        True if successfully revoked, False if not found.

    Raises:
        NotImplementedError: If this gateway does not support revocation.
    """
    raise NotImplementedError(
        f"{type(self).__name__} does not support revoke"
    )

MockGateway

paygraph.gateways.mock.MockGateway

Bases: BaseGateway

Mock card gateway for development and testing.

Generates fake card numbers. When auto_approve is False (default), prompts for human approval in the terminal before issuing a card.

Source code in src/paygraph/gateways/mock.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
class MockGateway(BaseGateway):
    """Mock card gateway for development and testing.

    Generates fake card numbers. When ``auto_approve`` is False (default),
    prompts for human approval in the terminal before issuing a card.
    """

    def __init__(self, auto_approve: bool = False) -> None:
        """Initialize the mock gateway.

        Args:
            auto_approve: If True, skip the terminal approval prompt and
                approve all requests automatically.
        """
        self.auto_approve = auto_approve
        self._cards: dict[str, CardResult] = {}

    def execute(self, amount_cents: int, vendor: str, memo: str) -> CardResult:
        """Create a mock virtual card, optionally prompting for approval.

        Args:
            amount_cents: Spend limit in cents.
            vendor: Name of the vendor.
            memo: Justification for the spend.

        Returns:
            A ``CardResult`` with a fake PAN (``4111111111111111``).

        Raises:
            SpendDeniedError: If the human denies the approval prompt.
        """
        amount_dollars = amount_cents / 100
        if not self.auto_approve:
            response = input(
                f"[PayGraph] Approve ${amount_dollars:.2f} for {vendor}? [Y/n]: "
            )
            if response.strip().lower() not in ("", "y", "yes"):
                raise SpendDeniedError(
                    f"Human denied spend of ${amount_dollars:.2f} for {vendor}"
                )

        token = f"mock_{secrets.token_hex(8)}"
        card = CardResult(
            pan="4111111111111111",
            cvv="123",
            expiry="12/28",
            spend_limit_cents=amount_cents,
            amount_cents=amount_cents,
            gateway_ref=token,
            gateway_type="mock",
        )
        self._cards[token] = card
        return card

    def revoke(self, gateway_ref: str) -> bool:
        """Remove a mock card from the internal store.

        Args:
            gateway_ref: The ``gateway_ref`` of the card to revoke.

        Returns:
            True if the card existed and was removed, False otherwise.
        """
        return self._cards.pop(gateway_ref, None) is not None

__init__(auto_approve=False)

Initialize the mock gateway.

Parameters:

Name Type Description Default
auto_approve bool

If True, skip the terminal approval prompt and approve all requests automatically.

 False
Source code in src/paygraph/gateways/mock.py 
14
15
16
17
18
19
20
21
22
def __init__(self, auto_approve: bool = False) -> None:
    """Initialize the mock gateway.

    Args:
        auto_approve: If True, skip the terminal approval prompt and
            approve all requests automatically.
    """
    self.auto_approve = auto_approve
    self._cards: dict[str, CardResult] = {}

execute(amount_cents, vendor, memo)

Create a mock virtual card, optionally prompting for approval.

Parameters:

Name Type Description Default
amount_cents int

Spend limit in cents.

 required
vendor str

Name of the vendor.

 required
memo str

Justification for the spend.

 required

Returns:

Type Description
CardResult

A CardResult with a fake PAN (4111111111111111).

Raises:

Type Description
SpendDeniedError

If the human denies the approval prompt.
Source code in src/paygraph/gateways/mock.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
def execute(self, amount_cents: int, vendor: str, memo: str) -> CardResult:
    """Create a mock virtual card, optionally prompting for approval.

    Args:
        amount_cents: Spend limit in cents.
        vendor: Name of the vendor.
        memo: Justification for the spend.

    Returns:
        A ``CardResult`` with a fake PAN (``4111111111111111``).

    Raises:
        SpendDeniedError: If the human denies the approval prompt.
    """
    amount_dollars = amount_cents / 100
    if not self.auto_approve:
        response = input(
            f"[PayGraph] Approve ${amount_dollars:.2f} for {vendor}? [Y/n]: "
        )
        if response.strip().lower() not in ("", "y", "yes"):
            raise SpendDeniedError(
                f"Human denied spend of ${amount_dollars:.2f} for {vendor}"
            )

    token = f"mock_{secrets.token_hex(8)}"
    card = CardResult(
        pan="4111111111111111",
        cvv="123",
        expiry="12/28",
        spend_limit_cents=amount_cents,
        amount_cents=amount_cents,
        gateway_ref=token,
        gateway_type="mock",
    )
    self._cards[token] = card
    return card

revoke(gateway_ref)

Remove a mock card from the internal store.

Parameters:

Name Type Description Default
gateway_ref str

The gateway_ref of the card to revoke.

 required

Returns:

Type Description
bool

True if the card existed and was removed, False otherwise.
Source code in src/paygraph/gateways/mock.py 
61
62
63
64
65
66
67
68
69
70
def revoke(self, gateway_ref: str) -> bool:
    """Remove a mock card from the internal store.

    Args:
        gateway_ref: The ``gateway_ref`` of the card to revoke.

    Returns:
        True if the card existed and was removed, False otherwise.
    """
    return self._cards.pop(gateway_ref, None) is not None

StripeCardGateway

paygraph.gateways.stripe.StripeCardGateway

Bases: BaseGateway

Stripe Issuing gateway that creates real virtual cards.

Automatically detects test vs live mode from the API key prefix (sk_test_ or sk_live_). Creates or reuses a Stripe cardholder.

Parameters:

Name Type Description Default
api_key str

Stripe secret key (must start with sk_test_ or sk_live_).

 required
cardholder_id str | None

Existing Stripe cardholder ID to use. If None, one is created or reused automatically.

 None
currency str

ISO currency code (default "usd").

 'usd'
billing_address dict[str, str] | None

Cardholder billing address dict. Uses a San Francisco default if not provided.

 None
single_use bool

If True (default), a new card is created per transaction. If False, reuses the same card and updates the spending limit.

 True
allowed_mccs list[str] | None

Stripe MCC allowlist applied at the card level.

 None
blocked_mccs list[str] | None

Stripe MCC blocklist applied at the card level.

 None
Source code in src/paygraph/gateways/stripe.py 
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
class StripeCardGateway(BaseGateway):
    """Stripe Issuing gateway that creates real virtual cards.

    Automatically detects test vs live mode from the API key prefix
    (``sk_test_`` or ``sk_live_``). Creates or reuses a Stripe cardholder.

    Args:
        api_key: Stripe secret key (must start with ``sk_test_`` or ``sk_live_``).
        cardholder_id: Existing Stripe cardholder ID to use. If None, one is
            created or reused automatically.
        currency: ISO currency code (default ``"usd"``).
        billing_address: Cardholder billing address dict. Uses a San Francisco
            default if not provided.
        single_use: If True (default), a new card is created per transaction.
            If False, reuses the same card and updates the spending limit.
        allowed_mccs: Stripe MCC allowlist applied at the card level.
        blocked_mccs: Stripe MCC blocklist applied at the card level.
    """

    API_BASE = "https://api.stripe.com"

    def __init__(
        self,
        api_key: str,
        cardholder_id: str | None = None,
        currency: str = "usd",
        billing_address: dict[str, str] | None = None,
        single_use: bool = True,
        allowed_mccs: list[str] | None = None,
        blocked_mccs: list[str] | None = None,
    ):
        if api_key.startswith("sk_test_"):
            self._gateway_type = "stripe_test"
        elif api_key.startswith("sk_live_"):
            self._gateway_type = "stripe_live"
        else:
            raise GatewayError(
                "Invalid Stripe API key — must start with sk_test_ or sk_live_"
            )

        self._client = httpx.Client(
            base_url=self.API_BASE,
            headers={"Authorization": f"Bearer {api_key}"},
            timeout=30,
        )
        self._cardholder_id = cardholder_id
        self._currency = currency.lower()
        self._billing = billing_address or _DEFAULT_BILLING
        self._single_use = single_use
        self._allowed_mccs = allowed_mccs
        self._blocked_mccs = blocked_mccs
        self._card_cache: dict[str, str] | None = (
            None  # cached card detail for reuse mode
        )

    def _find_existing_cardholder(self) -> str | None:
        """Look for an existing PayGraph Agent cardholder."""
        try:
            resp = self._client.get(
                "/v1/issuing/cardholders",
                params={"email": "agent@paygraph.dev", "status": "active", "limit": 1},
            )
            resp.raise_for_status()
        except httpx.HTTPError:
            return None

        data = resp.json().get("data", [])
        if data:
            return data[0]["id"]
        return None

    def _ensure_cardholder(self) -> str:
        if self._cardholder_id:
            return self._cardholder_id

        # Reuse existing cardholder if one exists
        existing = self._find_existing_cardholder()
        if existing:
            self._cardholder_id = existing
            return self._cardholder_id

        try:
            resp = self._client.post(
                "/v1/issuing/cardholders",
                data={
                    "type": "individual",
                    "name": "PayGraph Agent",
                    "email": "agent@paygraph.dev",
                    "status": "active",
                    "individual[first_name]": "PayGraph",
                    "individual[last_name]": "Agent",
                    "phone_number": "+18005550000",
                    **{f"billing[address][{k}]": v for k, v in self._billing.items()},
                },
            )
            resp.raise_for_status()
        except httpx.HTTPError as e:
            raise _map_stripe_error(e) from e

        self._cardholder_id = resp.json()["id"]
        return self._cardholder_id

    def _create_card(
        self, cardholder_id: str, amount_cents: int, vendor: str, memo: str
    ) -> dict:
        """Create a new Stripe Issuing card and return its expanded detail."""
        card_data: dict[str, str] = {
            "type": "virtual",
            "cardholder": cardholder_id,
            "currency": self._currency,
            "status": "active",
            "spending_controls[spending_limits][0][amount]": str(amount_cents),
            "spending_controls[spending_limits][0][interval]": "all_time",
        }
        if self._allowed_mccs:
            for i, mcc in enumerate(self._allowed_mccs):
                card_data[f"spending_controls[allowed_categories][{i}]"] = mcc
        if self._blocked_mccs:
            for i, mcc in enumerate(self._blocked_mccs):
                card_data[f"spending_controls[blocked_categories][{i}]"] = mcc
        if memo:
            card_data["metadata[memo]"] = memo[:500]
        if vendor:
            card_data["metadata[vendor]"] = vendor[:100]

        try:
            resp = self._client.post("/v1/issuing/cards", data=card_data)
            resp.raise_for_status()
        except httpx.HTTPError as e:
            raise _map_stripe_error(e) from e

        card_id = resp.json()["id"]
        return self._get_card_detail(card_id)

    def _get_card_detail(self, card_id: str) -> dict:
        """Fetch expanded card detail (PAN, CVC)."""
        try:
            resp = self._client.get(
                f"/v1/issuing/cards/{card_id}",
                params={"expand[]": ["number", "cvc"]},
            )
            resp.raise_for_status()
        except httpx.HTTPError as e:
            raise _map_stripe_error(e) from e
        return resp.json()

    def _update_spending_limit(self, card_id: str, amount_cents: int) -> None:
        """Update the spending limit on an existing card."""
        try:
            resp = self._client.post(
                f"/v1/issuing/cards/{card_id}",
                data={
                    "spending_controls[spending_limits][0][amount]": str(amount_cents),
                    "spending_controls[spending_limits][0][interval]": "all_time",
                },
            )
            resp.raise_for_status()
        except httpx.HTTPError as e:
            raise _map_stripe_error(e) from e

    def execute(self, amount_cents: int, vendor: str, memo: str) -> CardResult:
        """Create a Stripe Issuing virtual card with the given spend limit.

        Calls the Stripe ``/v1/issuing/cards`` API to mint a new card
        (or update the existing card's limit if ``single_use`` is False).

        Args:
            amount_cents: Spend limit in cents.
            vendor: Name of the vendor (stored in card metadata).
            memo: Justification (stored in card metadata, truncated to 500 chars).

        Returns:
            A ``CardResult`` with the real PAN, CVV, and expiry.

        Raises:
            GatewayError: If the Stripe API call fails.
        """
        cardholder_id = self._ensure_cardholder()

        if self._single_use or self._card_cache is None:
            detail = self._create_card(cardholder_id, amount_cents, vendor, memo)
            if not self._single_use:
                self._card_cache = detail
        else:
            detail = self._card_cache
            self._update_spending_limit(detail["id"], amount_cents)

        exp_month = detail["exp_month"]
        exp_year = detail["exp_year"]

        return CardResult(
            pan=detail["number"],
            cvv=detail["cvc"],
            expiry=f"{exp_month:02d}/{exp_year % 100:02d}",
            spend_limit_cents=amount_cents,
            amount_cents=amount_cents,
            gateway_ref=detail["id"],
            gateway_type=self._gateway_type,
        )

    def revoke(self, gateway_ref: str) -> bool:
        """Cancel a Stripe Issuing card.

        Sets the card status to ``"canceled"`` via the Stripe API.

        Args:
            gateway_ref: The Stripe card ID (``ic_...``).

        Returns:
            True if the card was canceled, False if not found (404).

        Raises:
            GatewayError: If the Stripe API call fails (non-404).
        """
        try:
            resp = self._client.post(
                f"/v1/issuing/cards/{gateway_ref}",
                data={"status": "canceled"},
            )
            resp.raise_for_status()
        except httpx.HTTPStatusError as e:
            if e.response.status_code == 404:
                return False
            raise _map_stripe_error(e) from e
        except httpx.HTTPError as e:
            raise _map_stripe_error(e) from e
        return True

execute(amount_cents, vendor, memo)

Create a Stripe Issuing virtual card with the given spend limit.

Calls the Stripe /v1/issuing/cards API to mint a new card (or update the existing card's limit if single_use is False).

Parameters:

Name Type Description Default
amount_cents int

Spend limit in cents.

 required
vendor str

Name of the vendor (stored in card metadata).

 required
memo str

Justification (stored in card metadata, truncated to 500 chars).

 required

Returns:

Type Description
CardResult

A CardResult with the real PAN, CVV, and expiry.

Raises:

Type Description
GatewayError

If the Stripe API call fails.
Source code in src/paygraph/gateways/stripe.py 
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
def execute(self, amount_cents: int, vendor: str, memo: str) -> CardResult:
    """Create a Stripe Issuing virtual card with the given spend limit.

    Calls the Stripe ``/v1/issuing/cards`` API to mint a new card
    (or update the existing card's limit if ``single_use`` is False).

    Args:
        amount_cents: Spend limit in cents.
        vendor: Name of the vendor (stored in card metadata).
        memo: Justification (stored in card metadata, truncated to 500 chars).

    Returns:
        A ``CardResult`` with the real PAN, CVV, and expiry.

    Raises:
        GatewayError: If the Stripe API call fails.
    """
    cardholder_id = self._ensure_cardholder()

    if self._single_use or self._card_cache is None:
        detail = self._create_card(cardholder_id, amount_cents, vendor, memo)
        if not self._single_use:
            self._card_cache = detail
    else:
        detail = self._card_cache
        self._update_spending_limit(detail["id"], amount_cents)

    exp_month = detail["exp_month"]
    exp_year = detail["exp_year"]

    return CardResult(
        pan=detail["number"],
        cvv=detail["cvc"],
        expiry=f"{exp_month:02d}/{exp_year % 100:02d}",
        spend_limit_cents=amount_cents,
        amount_cents=amount_cents,
        gateway_ref=detail["id"],
        gateway_type=self._gateway_type,
    )

revoke(gateway_ref)

Cancel a Stripe Issuing card.

Sets the card status to "canceled" via the Stripe API.

Parameters:

Name Type Description Default
gateway_ref str

The Stripe card ID (ic_...).

 required

Returns:

Type Description
bool

True if the card was canceled, False if not found (404).

Raises:

Type Description
GatewayError

If the Stripe API call fails (non-404).
Source code in src/paygraph/gateways/stripe.py 
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
def revoke(self, gateway_ref: str) -> bool:
    """Cancel a Stripe Issuing card.

    Sets the card status to ``"canceled"`` via the Stripe API.

    Args:
        gateway_ref: The Stripe card ID (``ic_...``).

    Returns:
        True if the card was canceled, False if not found (404).

    Raises:
        GatewayError: If the Stripe API call fails (non-404).
    """
    try:
        resp = self._client.post(
            f"/v1/issuing/cards/{gateway_ref}",
            data={"status": "canceled"},
        )
        resp.raise_for_status()
    except httpx.HTTPStatusError as e:
        if e.response.status_code == 404:
            return False
        raise _map_stripe_error(e) from e
    except httpx.HTTPError as e:
        raise _map_stripe_error(e) from e
    return True

X402Receipt

paygraph.gateways.x402.X402Receipt = X402Result module-attribute

X402Gateway

paygraph.gateways.x402.X402Gateway

Bases: BaseGateway

Gateway for x402 HTTP 402 payments using on-chain USDC.

Supports EVM chains (Base, Ethereum, etc.) and Solana. At least one private key must be provided. The x402 SDK is imported lazily — it only needs to be installed when this gateway is used.

Parameters:

Name Type Description Default
evm_private_key str | None

Hex-encoded EVM private key (e.g. "0x...").

 None
svm_private_key str | None

Base58-encoded Solana private key.

 None
Source code in src/paygraph/gateways/x402.py 
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
class X402Gateway(BaseGateway):
    """Gateway for x402 HTTP 402 payments using on-chain USDC.

    Supports EVM chains (Base, Ethereum, etc.) and Solana. At least one
    private key must be provided. The x402 SDK is imported lazily — it
    only needs to be installed when this gateway is used.

    Args:
        evm_private_key: Hex-encoded EVM private key (e.g. ``"0x..."``).
        svm_private_key: Base58-encoded Solana private key.
    """

    def __init__(
        self,
        evm_private_key: str | None = None,
        svm_private_key: str | None = None,
    ) -> None:
        if not evm_private_key and not svm_private_key:
            raise ValueError(
                "At least one of evm_private_key or svm_private_key is required"
            )

        try:
            from x402 import x402Client
        except ImportError:
            raise ImportError(
                "x402 integration requires the x402 SDK. "
                "Install it with: pip install paygraph[x402]"
            )

        self._client = x402Client()

        if evm_private_key:
            from eth_account import Account
            from x402.mechanisms.evm import EthAccountSigner
            from x402.mechanisms.evm.exact.register import register_exact_evm_client

            account = Account.from_key(evm_private_key)
            register_exact_evm_client(self._client, EthAccountSigner(account))
            self._payer = account.address

        if svm_private_key:
            from x402.mechanisms.svm import KeypairSigner
            from x402.mechanisms.svm.exact.register import register_exact_svm_client

            svm_signer = KeypairSigner.from_base58(svm_private_key)
            register_exact_svm_client(self._client, svm_signer)
            if not evm_private_key:
                self._payer = str(svm_signer.address)

    async def execute_async(
        self,
        amount_cents: int,
        vendor: str,
        memo: str,
        *,
        url: str = "",
        method: str = "GET",
        headers: dict | None = None,
        body: str | None = None,
    ) -> X402Result:
        """Make a paid HTTP request via the x402 protocol (async).

        Use this coroutine directly from async contexts such as LangGraph
        agents, FastAPI handlers, or Jupyter notebooks where an event loop
        is already running. Calling the synchronous :meth:`execute`
        from those contexts will raise ``RuntimeError``.

        Args:
            amount_cents: Payment amount in cents.
            vendor: Name of the vendor or service.
            memo: Justification or memo for the payment.
            url: The x402-enabled endpoint URL.
            method: HTTP method (default ``"GET"``).
            headers: Optional additional HTTP headers.
            body: Optional request body string.

        Returns:
            An ``X402Result`` with the response body and transaction details.

        Raises:
            RuntimeError: If the x402 payment fails (still 402 after retry).
        """

        from x402.http.clients import x402HttpxClient

        async with x402HttpxClient(self._client) as http:
            req_kwargs: dict = {}
            if headers:
                req_kwargs["headers"] = headers
            if body:
                req_kwargs["content"] = body

            response = await http.request(method, url, **req_kwargs)
            await response.aread()

            # If still 402 after the SDK's retry, payment failed
            if response.status_code == 402:
                try:
                    error_data = response.json()
                    error_msg = error_data.get("error", "Payment required but failed")
                except Exception:
                    error_msg = "Payment required but failed"
                raise RuntimeError(f"x402 payment failed: {error_msg}")

            tx_hash = ""
            network = ""

            # Parse the base64-encoded payment-response header
            payment_header = response.headers.get("payment-response", "")
            if payment_header:
                try:
                    settle = json.loads(base64.b64decode(payment_header))
                    tx_hash = settle.get("transaction", "") or settle.get(
                        "transactionId", ""
                    )
                    network = settle.get("network", "")
                except Exception:
                    pass

            return X402Result(
                url=url,
                amount_cents=amount_cents,
                network=network,
                transaction_hash=tx_hash,
                payer=self._payer,
                gateway_ref=tx_hash or f"x402_{id(response)}",
                gateway_type="x402",
                status_code=response.status_code,
                response_body=response.text,
                content_type=response.headers.get("content-type", ""),
            )

    def execute(
        self,
        amount_cents: int,
        vendor: str,
        memo: str,
        *,
        url: str = "",
        method: str = "GET",
        headers: dict | None = None,
        body: str | None = None,
    ) -> X402Result:
        """Make a paid HTTP request via the x402 protocol (sync).

        Synchronous wrapper around :meth:`execute_async`. Safe to call
        from any context — including scripts, CLI, and environments with a
        running event loop (LangGraph agents, FastAPI handlers, Jupyter
        notebooks). When a loop is already running in the current thread, the
        call is automatically dispatched to a worker thread with its own fresh
        event loop. For fully-async callers that prefer to avoid the thread
        overhead, ``await gateway.execute_async(...)`` is also available.

        Args:
            amount_cents: Payment amount in cents.
            vendor: Name of the vendor or service.
            memo: Justification or memo for the payment.
            url: The x402-enabled endpoint URL.
            method: HTTP method (default ``"GET"``).
            headers: Optional additional HTTP headers.
            body: Optional request body string.

        Returns:
            An ``X402Result`` with the response body and transaction details.

        Raises:
            RuntimeError: If the x402 payment fails (still 402 after retry).
        """
        coro = self.execute_async(
            amount_cents, vendor, memo, url=url, method=method, headers=headers, body=body
        )

        try:
            asyncio.get_running_loop()
            running = True
        except RuntimeError:
            running = False

        if running:
            # A loop is already running in this thread (LangGraph, FastAPI, Jupyter,
            # etc.). Dispatch to a worker thread so asyncio.run() can create a fresh
            # event loop there without interfering with the caller's loop.
            with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
                return pool.submit(asyncio.run, coro).result()

        return asyncio.run(coro)

execute_async(amount_cents, vendor, memo, *, url='', method='GET', headers=None, body=None) async

Make a paid HTTP request via the x402 protocol (async).

Use this coroutine directly from async contexts such as LangGraph agents, FastAPI handlers, or Jupyter notebooks where an event loop is already running. Calling the synchronous :meth:execute from those contexts will raise RuntimeError.

Parameters:

Name Type Description Default
amount_cents int

Payment amount in cents.

 required
vendor str

Name of the vendor or service.

 required
memo str

Justification or memo for the payment.

 required
url str

The x402-enabled endpoint URL.

 ''
method str

HTTP method (default "GET").

 'GET'
headers dict | None

Optional additional HTTP headers.

 None
body str | None

Optional request body string.

 None

Returns:

Type Description
X402Result

An X402Result with the response body and transaction details.

Raises:

Type Description
RuntimeError

If the x402 payment fails (still 402 after retry).
Source code in src/paygraph/gateways/x402.py 
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
async def execute_async(
    self,
    amount_cents: int,
    vendor: str,
    memo: str,
    *,
    url: str = "",
    method: str = "GET",
    headers: dict | None = None,
    body: str | None = None,
) -> X402Result:
    """Make a paid HTTP request via the x402 protocol (async).

    Use this coroutine directly from async contexts such as LangGraph
    agents, FastAPI handlers, or Jupyter notebooks where an event loop
    is already running. Calling the synchronous :meth:`execute`
    from those contexts will raise ``RuntimeError``.

    Args:
        amount_cents: Payment amount in cents.
        vendor: Name of the vendor or service.
        memo: Justification or memo for the payment.
        url: The x402-enabled endpoint URL.
        method: HTTP method (default ``"GET"``).
        headers: Optional additional HTTP headers.
        body: Optional request body string.

    Returns:
        An ``X402Result`` with the response body and transaction details.

    Raises:
        RuntimeError: If the x402 payment fails (still 402 after retry).
    """

    from x402.http.clients import x402HttpxClient

    async with x402HttpxClient(self._client) as http:
        req_kwargs: dict = {}
        if headers:
            req_kwargs["headers"] = headers
        if body:
            req_kwargs["content"] = body

        response = await http.request(method, url, **req_kwargs)
        await response.aread()

        # If still 402 after the SDK's retry, payment failed
        if response.status_code == 402:
            try:
                error_data = response.json()
                error_msg = error_data.get("error", "Payment required but failed")
            except Exception:
                error_msg = "Payment required but failed"
            raise RuntimeError(f"x402 payment failed: {error_msg}")

        tx_hash = ""
        network = ""

        # Parse the base64-encoded payment-response header
        payment_header = response.headers.get("payment-response", "")
        if payment_header:
            try:
                settle = json.loads(base64.b64decode(payment_header))
                tx_hash = settle.get("transaction", "") or settle.get(
                    "transactionId", ""
                )
                network = settle.get("network", "")
            except Exception:
                pass

        return X402Result(
            url=url,
            amount_cents=amount_cents,
            network=network,
            transaction_hash=tx_hash,
            payer=self._payer,
            gateway_ref=tx_hash or f"x402_{id(response)}",
            gateway_type="x402",
            status_code=response.status_code,
            response_body=response.text,
            content_type=response.headers.get("content-type", ""),
        )

execute(amount_cents, vendor, memo, *, url='', method='GET', headers=None, body=None)

Make a paid HTTP request via the x402 protocol (sync).

Synchronous wrapper around :meth:execute_async. Safe to call from any context — including scripts, CLI, and environments with a running event loop (LangGraph agents, FastAPI handlers, Jupyter notebooks). When a loop is already running in the current thread, the call is automatically dispatched to a worker thread with its own fresh event loop. For fully-async callers that prefer to avoid the thread overhead, await gateway.execute_async(...) is also available.

Parameters:

Name Type Description Default
amount_cents int

Payment amount in cents.

 required
vendor str

Name of the vendor or service.

 required
memo str

Justification or memo for the payment.

 required
url str

The x402-enabled endpoint URL.

 ''
method str

HTTP method (default "GET").

 'GET'
headers dict | None

Optional additional HTTP headers.

 None
body str | None

Optional request body string.

 None

Returns:

Type Description
X402Result

An X402Result with the response body and transaction details.

Raises:

Type Description
RuntimeError

If the x402 payment fails (still 402 after retry).
Source code in src/paygraph/gateways/x402.py 
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
def execute(
    self,
    amount_cents: int,
    vendor: str,
    memo: str,
    *,
    url: str = "",
    method: str = "GET",
    headers: dict | None = None,
    body: str | None = None,
) -> X402Result:
    """Make a paid HTTP request via the x402 protocol (sync).

    Synchronous wrapper around :meth:`execute_async`. Safe to call
    from any context — including scripts, CLI, and environments with a
    running event loop (LangGraph agents, FastAPI handlers, Jupyter
    notebooks). When a loop is already running in the current thread, the
    call is automatically dispatched to a worker thread with its own fresh
    event loop. For fully-async callers that prefer to avoid the thread
    overhead, ``await gateway.execute_async(...)`` is also available.

    Args:
        amount_cents: Payment amount in cents.
        vendor: Name of the vendor or service.
        memo: Justification or memo for the payment.
        url: The x402-enabled endpoint URL.
        method: HTTP method (default ``"GET"``).
        headers: Optional additional HTTP headers.
        body: Optional request body string.

    Returns:
        An ``X402Result`` with the response body and transaction details.

    Raises:
        RuntimeError: If the x402 payment fails (still 402 after retry).
    """
    coro = self.execute_async(
        amount_cents, vendor, memo, url=url, method=method, headers=headers, body=body
    )

    try:
        asyncio.get_running_loop()
        running = True
    except RuntimeError:
        running = False

    if running:
        # A loop is already running in this thread (LangGraph, FastAPI, Jupyter,
        # etc.). Dispatch to a worker thread so asyncio.run() can create a fresh
        # event loop there without interfering with the caller's loop.
        with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
            return pool.submit(asyncio.run, coro).result()

    return asyncio.run(coro)

MockX402Gateway

paygraph.gateways.mock_x402.MockX402Gateway

Bases: BaseGateway

Mock x402 gateway for testing without blockchain access.

Simulates the x402 payment flow by generating fake transaction hashes and returning configurable response bodies. Optionally prompts for human approval.

Parameters:

Name Type Description Default
auto_approve bool

If True, skip the terminal approval prompt.

 False
response_body str

Canned response body to return.

 '{}'
status_code int

HTTP status code of the mock response.

 200
content_type str

Content-Type of the mock response.

 'application/json'
Source code in src/paygraph/gateways/mock_x402.py 
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
class MockX402Gateway(BaseGateway):
    """Mock x402 gateway for testing without blockchain access.

    Simulates the x402 payment flow by generating fake transaction hashes
    and returning configurable response bodies. Optionally prompts for
    human approval.

    Args:
        auto_approve: If True, skip the terminal approval prompt.
        response_body: Canned response body to return.
        status_code: HTTP status code of the mock response.
        content_type: Content-Type of the mock response.
    """

    def __init__(
        self,
        auto_approve: bool = False,
        response_body: str = "{}",
        status_code: int = 200,
        content_type: str = "application/json",
    ) -> None:
        self.auto_approve = auto_approve
        self.response_body = response_body
        self.status_code = status_code
        self.content_type = content_type
        self._receipts: dict[str, X402Result] = {}

    async def execute_async(
        self,
        amount_cents: int,
        vendor: str,
        memo: str,
        *,
        url: str = "",
        method: str = "GET",
        headers: dict | None = None,
        body: str | None = None,
    ) -> X402Result:
        """Async variant — delegates to :meth:`execute`."""
        return self.execute(
            amount_cents, vendor, memo, url=url, method=method, headers=headers, body=body
        )

    def execute(
        self,
        amount_cents: int,
        vendor: str,
        memo: str,
        *,
        url: str = "",
        method: str = "GET",
        headers: dict | None = None,
        body: str | None = None,
    ) -> X402Result:
        """Simulate an x402 payment, optionally prompting for approval.

        Args:
            amount_cents: Payment amount in cents.
            vendor: Name of the vendor.
            memo: Justification for the payment.
            url: The x402-enabled endpoint URL.
            method: HTTP method (ignored in mock).
            headers: Optional HTTP headers (ignored in mock).
            body: Optional request body (ignored in mock).

        Returns:
            An ``X402Result`` with a fake transaction hash and the
            configured response body.

        Raises:
            SpendDeniedError: If the human denies the approval prompt.
        """
        amount_dollars = amount_cents / 100
        if not self.auto_approve:
            response = input(
                f"[PayGraph] Approve x402 payment ${amount_dollars:.2f} for {vendor}? [Y/n]: "
            )
            if response.strip().lower() not in ("", "y", "yes"):
                raise SpendDeniedError(
                    f"Human denied x402 payment of ${amount_dollars:.2f} for {vendor}"
                )

        tx_hash = f"0xmock_{secrets.token_hex(16)}"
        receipt = X402Result(
            url=url,
            amount_cents=amount_cents,
            network="eip155:8453",
            transaction_hash=tx_hash,
            payer="0xMockPayer1234567890abcdef",
            gateway_ref=tx_hash,
            gateway_type="x402",
            status_code=self.status_code,
            response_body=self.response_body,
            content_type=self.content_type,
        )
        self._receipts[tx_hash] = receipt
        return receipt

execute_async(amount_cents, vendor, memo, *, url='', method='GET', headers=None, body=None) async

Async variant — delegates to :meth:execute.

Source code in src/paygraph/gateways/mock_x402.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
async def execute_async(
    self,
    amount_cents: int,
    vendor: str,
    memo: str,
    *,
    url: str = "",
    method: str = "GET",
    headers: dict | None = None,
    body: str | None = None,
) -> X402Result:
    """Async variant — delegates to :meth:`execute`."""
    return self.execute(
        amount_cents, vendor, memo, url=url, method=method, headers=headers, body=body
    )

execute(amount_cents, vendor, memo, *, url='', method='GET', headers=None, body=None)

Simulate an x402 payment, optionally prompting for approval.

Parameters:

Name Type Description Default
amount_cents int

Payment amount in cents.

 required
vendor str

Name of the vendor.

 required
memo str

Justification for the payment.

 required
url str

The x402-enabled endpoint URL.

 ''
method str

HTTP method (ignored in mock).

 'GET'
headers dict | None

Optional HTTP headers (ignored in mock).

 None
body str | None

Optional request body (ignored in mock).

 None

Returns:

Type Description
X402Result

An X402Result with a fake transaction hash and the
X402Result

configured response body.

Raises:

Type Description
SpendDeniedError

If the human denies the approval prompt.
Source code in src/paygraph/gateways/mock_x402.py 
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def execute(
    self,
    amount_cents: int,
    vendor: str,
    memo: str,
    *,
    url: str = "",
    method: str = "GET",
    headers: dict | None = None,
    body: str | None = None,
) -> X402Result:
    """Simulate an x402 payment, optionally prompting for approval.

    Args:
        amount_cents: Payment amount in cents.
        vendor: Name of the vendor.
        memo: Justification for the payment.
        url: The x402-enabled endpoint URL.
        method: HTTP method (ignored in mock).
        headers: Optional HTTP headers (ignored in mock).
        body: Optional request body (ignored in mock).

    Returns:
        An ``X402Result`` with a fake transaction hash and the
        configured response body.

    Raises:
        SpendDeniedError: If the human denies the approval prompt.
    """
    amount_dollars = amount_cents / 100
    if not self.auto_approve:
        response = input(
            f"[PayGraph] Approve x402 payment ${amount_dollars:.2f} for {vendor}? [Y/n]: "
        )
        if response.strip().lower() not in ("", "y", "yes"):
            raise SpendDeniedError(
                f"Human denied x402 payment of ${amount_dollars:.2f} for {vendor}"
            )

    tx_hash = f"0xmock_{secrets.token_hex(16)}"
    receipt = X402Result(
        url=url,
        amount_cents=amount_cents,
        network="eip155:8453",
        transaction_hash=tx_hash,
        payer="0xMockPayer1234567890abcdef",
        gateway_ref=tx_hash,
        gateway_type="x402",
        status_code=self.status_code,
        response_body=self.response_body,
        content_type=self.content_type,
    )
    self._receipts[tx_hash] = receipt
    return receipt