How an MCP agent pays for tools with x402 (TypeScript and Python)

import { createx402MCPClient } from "@x402/mcp";
import { ExactEvmScheme } from "@x402/evm/exact/client";
import { toClientEvmSigner } from "@x402/evm";
import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js";
import { privateKeyToAccount } from "viem/accounts";

const signer = toClientEvmSigner(privateKeyToAccount(process.env.AGENT_KEY));
const client = createx402MCPClient({
  name: "my-agent", version: "1.0.0",
  schemes: [{ network: "eip155:8453", client: new ExactEvmScheme(signer) }], // USDC on Base
  autoPayment: true,
  onPaymentRequested: async ({ paymentRequired }) => {
    const base = paymentRequired.accepts.find((a) => a.network === "eip155:8453");
    return Number(base?.amount) <= 50_000; // pay at most USD 0.05 a call (USDC has 6 decimals)
  },
});
await client.connect(new StreamableHTTPClientTransport(new URL("https://tanod.dev/mcp")));
const result = await client.callTool("pdf_ocr", { url: "https://example.com/scan.pdf" });
console.log(result.content, result.paymentMade);

import asyncio
import os
from types import SimpleNamespace

from eth_account import Account
from mcp import ClientSession
from mcp.client.streamable_http import streamable_http_client
from x402.mcp import create_x402_mcp_client_from_config
from x402.mechanisms.evm.exact import ExactEvmClientScheme

BASE = "eip155:8453"  # USDC on Base
MAX_PER_CALL = 50_000  # USD 0.05 (USDC has 6 decimals)


class SessionAdapter:
    """Gives x402's client the call_tool(params) shape it expects."""

    def __init__(self, session):
        self._s = session

    async def call_tool(self, params, read_timeout_seconds=None, **_):
        if read_timeout_seconds is not None:  # x402 passes a timedelta, mcp 2.x wants seconds
            read_timeout_seconds = read_timeout_seconds.total_seconds()
        r = await self._s.call_tool(
            params["name"],
            params.get("arguments"),
            read_timeout_seconds=read_timeout_seconds,
            meta=params.get("_meta"),
        )
        d = r.model_dump(by_alias=True, exclude_none=True)
        return SimpleNamespace(
            content=d.get("content", []),
            isError=d.get("isError", False),
            structuredContent=d.get("structuredContent"),
            _meta=d.get("_meta", {}),
        )

    def __getattr__(self, name):
        return getattr(self._s, name)


def approve(ctx):
    base = next((a for a in ctx.payment_required.accepts if a.network == BASE), None)
    return base is not None and int(base.amount) <= MAX_PER_CALL


async def main():
    account = Account.from_key(os.environ["AGENT_KEY"])
    async with streamable_http_client("https://tanod.dev/mcp") as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            client = create_x402_mcp_client_from_config(
                SessionAdapter(session),
                {
                    "schemes": [{"network": BASE, "client": ExactEvmClientScheme(account)}],
                    "auto_payment": True,
                    "on_payment_requested": approve,
                },
            )
            result = await client.call_tool("pdf_ocr", {"url": "https://example.com/scan.pdf"})
            print(result.content, result.payment_made)


asyncio.run(main())