Skip to content

Gateways

BaseGateway

paygraph.gateways.base.BaseGateway

Bases: ABC

Abstract base class for card payment gateways.

Subclass this to implement a custom card gateway. You must implement execute_spend() and revoke().

Source code in src/paygraph/gateways/base.py 
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
class BaseGateway(ABC):
    """Abstract base class for card payment gateways.

    Subclass this to implement a custom card gateway. You must implement
    ``execute_spend()`` and ``revoke()``.
    """

    @abstractmethod
    def execute_spend(self, amount_cents: int, vendor: str, memo: str) -> VirtualCard:
        """Create a virtual card for the given spend.

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

        Returns:
            A ``VirtualCard`` with the card details.
        """
        ...

    @abstractmethod
    def revoke(self, gateway_ref: str) -> bool:
        """Revoke (cancel) a previously issued virtual card.

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

        Returns:
            True if the card was successfully revoked, False if not found.
        """
        ...

execute_spend(amount_cents, vendor, memo) abstractmethod

Create a virtual card for the given spend.

Parameters:

Name Type Description Default
amount_cents int

Spend limit in cents.

 required
vendor str

Name of the vendor.

 required
memo str

Justification or memo for the spend.

 required

Returns:

Type Description
VirtualCard

A VirtualCard with the card details.
Source code in src/paygraph/gateways/base.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
@abstractmethod
def execute_spend(self, amount_cents: int, vendor: str, memo: str) -> VirtualCard:
    """Create a virtual card for the given spend.

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

    Returns:
        A ``VirtualCard`` with the card details.
    """
    ...

revoke(gateway_ref) abstractmethod

Revoke (cancel) a previously issued virtual card.

Parameters:

Name Type Description Default
gateway_ref str

The gateway_ref from the VirtualCard.

 required

Returns:

Type Description
bool

True if the card was successfully revoked, False if not found.
Source code in src/paygraph/gateways/base.py 
62
63
64
65
66
67
68
69
70
71
72
@abstractmethod
def revoke(self, gateway_ref: str) -> bool:
    """Revoke (cancel) a previously issued virtual card.

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

    Returns:
        True if the card was successfully revoked, False if not found.
    """
    ...

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
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, VirtualCard] = {}

    def execute_spend(self, amount_cents: int, vendor: str, memo: str) -> VirtualCard:
        """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 ``VirtualCard`` 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 = VirtualCard(
            pan="4111111111111111",
            cvv="123",
            expiry="12/28",
            spend_limit_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, VirtualCard] = {}

execute_spend(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
VirtualCard

A VirtualCard 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
def execute_spend(self, amount_cents: int, vendor: str, memo: str) -> VirtualCard:
    """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 ``VirtualCard`` 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 = VirtualCard(
        pan="4111111111111111",
        cvv="123",
        expiry="12/28",
        spend_limit_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 
60
61
62
63
64
65
66
67
68
69
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 
 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
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
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.HTTPStatusError as e:
            msg = e.response.json().get("error", {}).get("message", str(e))
            raise GatewayError(f"Stripe API error: {msg}") from e
        except httpx.HTTPError as e:
            raise GatewayError(f"Stripe API unreachable: {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.HTTPStatusError as e:
            msg = e.response.json().get("error", {}).get("message", str(e))
            raise GatewayError(f"Stripe API error: {msg}") from e
        except httpx.HTTPError as e:
            raise GatewayError(f"Stripe API unreachable: {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.HTTPStatusError as e:
            msg = e.response.json().get("error", {}).get("message", str(e))
            raise GatewayError(f"Stripe API error: {msg}") from e
        except httpx.HTTPError as e:
            raise GatewayError(f"Stripe API unreachable: {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.HTTPStatusError as e:
            msg = e.response.json().get("error", {}).get("message", str(e))
            raise GatewayError(f"Stripe API error: {msg}") from e
        except httpx.HTTPError as e:
            raise GatewayError(f"Stripe API unreachable: {e}") from e

    def execute_spend(self, amount_cents: int, vendor: str, memo: str) -> VirtualCard:
        """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 ``VirtualCard`` 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 VirtualCard(
            pan=detail["number"],
            cvv=detail["cvc"],
            expiry=f"{exp_month:02d}/{exp_year % 100:02d}",
            spend_limit_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
            msg = e.response.json().get("error", {}).get("message", str(e))
            raise GatewayError(f"Stripe API error: {msg}") from e
        except httpx.HTTPError as e:
            raise GatewayError(f"Stripe API unreachable: {e}") from e
        return True

execute_spend(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
VirtualCard

A VirtualCard 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 
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
def execute_spend(self, amount_cents: int, vendor: str, memo: str) -> VirtualCard:
    """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 ``VirtualCard`` 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 VirtualCard(
        pan=detail["number"],
        cvv=detail["cvc"],
        expiry=f"{exp_month:02d}/{exp_year % 100:02d}",
        spend_limit_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 
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
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
        msg = e.response.json().get("error", {}).get("message", str(e))
        raise GatewayError(f"Stripe API error: {msg}") from e
    except httpx.HTTPError as e:
        raise GatewayError(f"Stripe API unreachable: {e}") from e
    return True

X402Receipt

paygraph.gateways.x402.X402Receipt dataclass

Result of a successful x402 payment.

Attributes:

Name Type Description
url str

The x402-enabled endpoint URL that was called.
amount_cents int

Amount paid in cents.
network str

Blockchain network identifier (e.g. "eip155:8453").
transaction_hash str

On-chain transaction hash.
payer str

Wallet address of the payer.
gateway_ref str

Unique reference (usually the transaction hash).
gateway_type str

Always "x402".
status_code int

HTTP status code of the response.
response_body str

Body of the HTTP response from the paid endpoint.
content_type str

Content-Type header of the response.
Source code in src/paygraph/gateways/x402.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
@dataclass
class X402Receipt:
    """Result of a successful x402 payment.

    Attributes:
        url: The x402-enabled endpoint URL that was called.
        amount_cents: Amount paid in cents.
        network: Blockchain network identifier (e.g. ``"eip155:8453"``).
        transaction_hash: On-chain transaction hash.
        payer: Wallet address of the payer.
        gateway_ref: Unique reference (usually the transaction hash).
        gateway_type: Always ``"x402"``.
        status_code: HTTP status code of the response.
        response_body: Body of the HTTP response from the paid endpoint.
        content_type: Content-Type header of the response.
    """

    url: str
    amount_cents: int
    network: str
    transaction_hash: str
    payer: str
    gateway_ref: str
    gateway_type: str = "x402"

    status_code: int = 200
    response_body: str = ""
    content_type: str = "application/json"

X402Gateway

paygraph.gateways.x402.X402Gateway

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 
 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
class X402Gateway:
    """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(
        self,
        url: str,
        amount_cents: int,
        vendor: str,
        memo: str,
        method: str = "GET",
        headers: dict | None = None,
        body: str | None = None,
    ) -> X402Receipt:
        from x402.http.clients import x402HttpxClient

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

            response = await http.request(method, url, **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 X402Receipt(
                url=url,
                amount_cents=amount_cents,
                network=network,
                transaction_hash=tx_hash,
                payer=self._payer,
                gateway_ref=tx_hash or f"x402_{id(response)}",
                status_code=response.status_code,
                response_body=response.text,
                content_type=response.headers.get("content-type", ""),
            )

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

        Synchronous wrapper around the async x402 client. Sends the
        request, handles the 402 payment challenge, and returns the
        final response.

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

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

        Raises:
            RuntimeError: If the x402 payment fails (still 402 after retry).
        """
        return asyncio.run(
            self._execute(url, amount_cents, vendor, memo, method, headers, body)
        )

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

Make a paid HTTP request via the x402 protocol.

Synchronous wrapper around the async x402 client. Sends the request, handles the 402 payment challenge, and returns the final response.

Parameters:

Name Type Description Default
url str

The x402-enabled endpoint URL.

 required
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
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
X402Receipt

An X402Receipt 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 
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
def execute_x402(
    self,
    url: str,
    amount_cents: int,
    vendor: str,
    memo: str,
    method: str = "GET",
    headers: dict | None = None,
    body: str | None = None,
) -> X402Receipt:
    """Make a paid HTTP request via the x402 protocol.

    Synchronous wrapper around the async x402 client. Sends the
    request, handles the 402 payment challenge, and returns the
    final response.

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

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

    Raises:
        RuntimeError: If the x402 payment fails (still 402 after retry).
    """
    return asyncio.run(
        self._execute(url, amount_cents, vendor, memo, method, headers, body)
    )

MockX402Gateway

paygraph.gateways.mock_x402.MockX402Gateway

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 
 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
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
class MockX402Gateway:
    """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, X402Receipt] = {}

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

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

        Returns:
            An ``X402Receipt`` 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 = X402Receipt(
            url=url,
            amount_cents=amount_cents,
            network="eip155:8453",
            transaction_hash=tx_hash,
            payer="0xMockPayer1234567890abcdef",
            gateway_ref=tx_hash,
            status_code=self.status_code,
            response_body=self.response_body,
            content_type=self.content_type,
        )
        self._receipts[tx_hash] = receipt
        return receipt

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

Simulate an x402 payment, optionally prompting for approval.

Parameters:

Name Type Description Default
url str

The endpoint URL.

 required
amount_cents int

Payment amount in cents.

 required
vendor str

Name of the vendor.

 required
memo str

Justification for the payment.

 required
method str

HTTP method (ignored in mock).

 'GET'
headers dict | None

Optional headers (ignored in mock).

 None
body str | None

Optional body (ignored in mock).

 None

Returns:

Type Description
X402Receipt

An X402Receipt with a fake transaction hash and the
X402Receipt

configured response body.

Raises:

Type Description
SpendDeniedError

If the human denies the approval prompt.
Source code in src/paygraph/gateways/mock_x402.py 
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
def execute_x402(
    self,
    url: str,
    amount_cents: int,
    vendor: str,
    memo: str,
    method: str = "GET",
    headers: dict | None = None,
    body: str | None = None,
) -> X402Receipt:
    """Simulate an x402 payment, optionally prompting for approval.

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

    Returns:
        An ``X402Receipt`` 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 = X402Receipt(
        url=url,
        amount_cents=amount_cents,
        network="eip155:8453",
        transaction_hash=tx_hash,
        payer="0xMockPayer1234567890abcdef",
        gateway_ref=tx_hash,
        status_code=self.status_code,
        response_body=self.response_body,
        content_type=self.content_type,
    )
    self._receipts[tx_hash] = receipt
    return receipt