feat: add EVM payment method (e.g. Base)

.changelog/brave-otters-swim.md

@@ -0,0 +1,5 @@
+---
+pympp: minor
+---
+
+Added `mpp.methods.evm`, a payment method for standard EVM chains (e.g. Base) that settles charges with a plain ERC-20 `transfer` instead of Tempo's native transaction type. Install with `pip install "pympp[evm]"`.

pyproject.toml

@@ -40,6 +40,10 @@ tempo = [
 stripe = [
     "pydantic>=2.0",
 ]
+evm = [
+    "eth-account>=0.11",
+    "eth-hash[pycryptodome]>=0.7",
+]
 server = ["pydantic>=2.0"]
 redis = ["redis>=5.0"]
 sqlite = ["aiosqlite>=0.20"]

src/mpp/methods/evm/__init__.py

@@ -0,0 +1,67 @@
+"""Generic EVM payment method (e.g. Base) for HTTP 402 authentication.
+
+pympp ships payment method implementations for Tempo (its own chain, using
+native account-abstraction transactions) and Stripe. This module is the
+equivalent for any standard EVM chain: it plugs into the same
+``Method``/``Intent`` protocols that ``mpp.methods.tempo`` implements, but
+moves value with a plain ERC-20 ``transfer`` instead of Tempo's native
+transaction type, since standard EVM chains (e.g. Base) have none of Tempo's
+extensions (memos, MACH fee tokens, escrow, access keys).
+
+Client side: builds and signs a standard EIP-1559 transaction calling
+``transfer(address,uint256)`` on the requested ERC-20 currency.
+
+Server side: broadcasts that signed transaction to an EVM JSON-RPC endpoint
+and confirms the mined receipt contains a matching ``Transfer`` log.
+
+Example:
+    # Client-side
+    from mpp.client import get
+    from mpp.methods.evm import evm, EvmAccount, ChargeIntent
+
+    account = EvmAccount.from_key("0x...")
+    response = await get(
+        "https://api.example.com/resource",
+        methods=[evm(
+            account=account,
+            intents={"charge": ChargeIntent()},
+        )],
+    )
+
+    # Server-side
+    from mpp.server import Mpp
+    from mpp.methods.evm import evm, ChargeIntent
+
+    server = Mpp.create(
+        method=evm(
+            recipient="0x...",
+            intents={"charge": ChargeIntent()},
+        ),
+    )
+"""
+
+from typing import Any
+
+from mpp._lazy_exports import load_lazy_attr
+from mpp.methods.evm._defaults import (
+    BASE_CHAIN_ID,
+    BASE_RPC_URL,
+    BASE_SEPOLIA_CHAIN_ID,
+    BASE_SEPOLIA_RPC_URL,
+    BASE_SEPOLIA_USDC,
+    BASE_USDC,
+    default_currency_for_chain,
+    rpc_url_for_chain,
+)
+
+_EXTRA_INSTALL_HINT = 'Install the "evm" extra to use this module: pip install "pympp[evm]"'
+
+_LAZY_EXPORTS = {
+    "mpp.methods.evm.account": ("EvmAccount",),
+    "mpp.methods.evm.client": ("EvmMethod", "TransactionError", "evm"),
+    "mpp.methods.evm.intents": ("ChargeIntent",),
+}
+
+
+def __getattr__(name: str) -> Any:
+    return load_lazy_attr(__name__, name, _LAZY_EXPORTS, globals(), _EXTRA_INSTALL_HINT)

src/mpp/methods/evm/__init__.pyi

@@ -0,0 +1,27 @@
+from mpp.methods.evm._defaults import BASE_CHAIN_ID as _BASE_CHAIN_ID
+from mpp.methods.evm._defaults import BASE_RPC_URL as _BASE_RPC_URL
+from mpp.methods.evm._defaults import BASE_SEPOLIA_CHAIN_ID as _BASE_SEPOLIA_CHAIN_ID
+from mpp.methods.evm._defaults import BASE_SEPOLIA_RPC_URL as _BASE_SEPOLIA_RPC_URL
+from mpp.methods.evm._defaults import BASE_SEPOLIA_USDC as _BASE_SEPOLIA_USDC
+from mpp.methods.evm._defaults import BASE_USDC as _BASE_USDC
+from mpp.methods.evm._defaults import default_currency_for_chain as _default_currency_for_chain
+from mpp.methods.evm._defaults import rpc_url_for_chain as _rpc_url_for_chain
+from mpp.methods.evm.account import EvmAccount as _EvmAccount
+from mpp.methods.evm.client import EvmMethod as _EvmMethod
+from mpp.methods.evm.client import TransactionError as _TransactionError
+from mpp.methods.evm.client import evm as _evm
+from mpp.methods.evm.intents import ChargeIntent as _ChargeIntent
+
+BASE_CHAIN_ID = _BASE_CHAIN_ID
+BASE_RPC_URL = _BASE_RPC_URL
+BASE_SEPOLIA_CHAIN_ID = _BASE_SEPOLIA_CHAIN_ID
+BASE_SEPOLIA_RPC_URL = _BASE_SEPOLIA_RPC_URL
+BASE_SEPOLIA_USDC = _BASE_SEPOLIA_USDC
+BASE_USDC = _BASE_USDC
+default_currency_for_chain = _default_currency_for_chain
+rpc_url_for_chain = _rpc_url_for_chain
+EvmAccount = _EvmAccount
+EvmMethod = _EvmMethod
+TransactionError = _TransactionError
+evm = _evm
+ChargeIntent = _ChargeIntent

src/mpp/methods/evm/_defaults.py

@@ -0,0 +1,53 @@
+"""Shared defaults for the EVM payment method (e.g. Base)."""
+
+from types import MappingProxyType
+
+BASE_CHAIN_ID = 8453
+BASE_SEPOLIA_CHAIN_ID = 84532
+
+BASE_RPC_URL = "https://mainnet.base.org"
+BASE_SEPOLIA_RPC_URL = "https://sepolia.base.org"
+
+# Native USDC on Base. See https://developer.coinbase.com/base/usdc
+BASE_USDC = "0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913"
+BASE_SEPOLIA_USDC = "0x036CbD53842c5426634e7929541eC2318f3dCF7e"
+
+CHAIN_RPC_URLS: MappingProxyType[int, str] = MappingProxyType(
+    {
+        BASE_CHAIN_ID: BASE_RPC_URL,
+        BASE_SEPOLIA_CHAIN_ID: BASE_SEPOLIA_RPC_URL,
+    }
+)
+
+DEFAULT_CURRENCIES: MappingProxyType[int, str] = MappingProxyType(
+    {
+        BASE_CHAIN_ID: BASE_USDC,
+        BASE_SEPOLIA_CHAIN_ID: BASE_SEPOLIA_USDC,
+    }
+)
+
+
+def rpc_url_for_chain(chain_id: int) -> str:
+    """Return the default RPC URL for a known EVM chain ID.
+
+    Raises:
+        ValueError: If the chain ID is not recognized.
+    """
+    url = CHAIN_RPC_URLS.get(chain_id)
+    if url is None:
+        raise ValueError(
+            f"Unknown chain_id {chain_id}. Known chains: {list(CHAIN_RPC_URLS)}. "
+            f"Pass rpc_url explicitly for custom chains."
+        )
+    return url
+
+
+def default_currency_for_chain(chain_id: int | None) -> str | None:
+    """Return the default USDC currency address for a known EVM chain ID.
+
+    Returns ``None`` for unknown chains and when ``chain_id`` is ``None`` — there
+    is no chain-agnostic default currency to fall back on for a custom chain.
+    """
+    if chain_id is None:
+        return None
+    return DEFAULT_CURRENCIES.get(chain_id)

src/mpp/methods/evm/_rpc.py

@@ -0,0 +1,32 @@
+"""Low-level JSON-RPC helper for the EVM payment method."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from mpp._defaults import DEFAULT_TIMEOUT
+
+
+async def _rpc_call(
+    rpc_url: str,
+    method: str,
+    params: list[Any],
+    *,
+    client: Any | None = None,
+) -> Any:
+    """Make a JSON-RPC call, raising ``RuntimeError`` on an RPC error."""
+    import httpx
+
+    payload = {"jsonrpc": "2.0", "method": method, "params": params, "id": 1}
+
+    if client is not None:
+        resp = await client.post(rpc_url, json=payload)
+    else:
+        async with httpx.AsyncClient(timeout=DEFAULT_TIMEOUT) as c:
+            resp = await c.post(rpc_url, json=payload)
+
+    resp.raise_for_status()
+    result = resp.json()
+    if "error" in result:
+        raise RuntimeError(f"RPC error: {result['error']}")
+    return result["result"]

src/mpp/methods/evm/account.py

@@ -0,0 +1,75 @@
+"""EVM account management for signing transactions.
+
+Wraps eth-account for key management and signing operations.
+"""
+
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass
+from pathlib import Path
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from eth_account.signers.local import LocalAccount
+
+
+@dataclass(frozen=True)
+class EvmAccount:
+    """Wrapper around eth-account for signing EVM transactions.
+
+    Example:
+        # From hex private key
+        account = EvmAccount.from_key("0x...")
+
+        # From environment variable
+        account = EvmAccount.from_env("EVM_PRIVATE_KEY")
+    """
+
+    _account: LocalAccount
+
+    @classmethod
+    def from_key(cls, private_key: str) -> EvmAccount:
+        """Load from hex private key (0x-prefixed)."""
+        from eth_account import Account
+
+        return cls(_account=Account.from_key(private_key))
+
+    @classmethod
+    def from_env(cls, var: str = "EVM_PRIVATE_KEY") -> EvmAccount:
+        """Load from environment variable.
+
+        Raises:
+            ValueError: If the environment variable is not set.
+        """
+        key = os.environ.get(var)
+        if not key:
+            raise ValueError(f"${var} not set")
+        return cls.from_key(key)
+
+    @classmethod
+    def from_file(cls, path: str) -> EvmAccount:
+        """Load from a local file containing a hex private key.
+
+        Args:
+            path: Path to the key file. The file's contents are stripped of
+                surrounding whitespace before use.
+
+        Raises:
+            ValueError: If the file is empty.
+            FileNotFoundError: If the file does not exist.
+        """
+        key = Path(path).read_text().strip()
+        if not key:
+            raise ValueError(f"{path} is empty")
+        return cls.from_key(key)
+
+    @property
+    def address(self) -> str:
+        """Get the account's Ethereum address."""
+        return self._account.address
+
+    @property
+    def private_key(self) -> str:
+        """Get the private key as a hex string for signing."""
+        return self._account.key.hex()