x402 Pay-Per-Call

import json, os, requests
from eth_account import Account
from x402 import x402ClientSync
from x402.http import encode_payment_signature_header, X_PAYMENT_HEADER
from x402.mechanisms.evm.exact import register_exact_evm_client
from x402.mechanisms.evm.signers import EthAccountSigner
from x402.schemas import PaymentRequired

ENDPOINT = "https://mcp.soundside.ai/mcp"

# Set up wallet + payment client
account = Account.from_key(os.environ["WALLET_PRIVATE_KEY"])  # 0x-prefixed hex
payment_client = x402ClientSync()
register_exact_evm_client(payment_client, EthAccountSigner(account))

def call_tool(tool: str, args: dict, timeout: int = 120) -> dict:
    """Call a Soundside MCP tool with automatic x402 payment."""
    session = requests.Session()

    # 1. Initialize MCP session (required for each call)
    init_r = session.post(ENDPOINT, json={
        "jsonrpc": "2.0", "id": "1", "method": "initialize",
        "params": {"protocolVersion": "2025-11-25", "capabilities": {},
                   "clientInfo": {"name": "x402-example", "version": "1.0"}},
    }, headers={"Content-Type": "application/json",
                "Accept": "application/json, text/event-stream"}, timeout=30)
    session_id = init_r.headers.get("mcp-session-id", "")

    headers = {"Content-Type": "application/json",
               "Accept": "application/json, text/event-stream",
               "mcp-session-id": session_id,
               "x402-wallet": account.address}  # required for resource attribution
    payload = {"jsonrpc": "2.0", "id": "2", "method": "tools/call",
               "params": {"name": tool, "arguments": args}}

    # 2. First attempt — server returns 402 Payment Required
    r = session.post(ENDPOINT, json=payload, headers=headers, timeout=timeout)

    if r.status_code == 402:
        # 3. Parse requirements from response BODY (not a header)
        pr = PaymentRequired.model_validate(r.json())
        # 4. Sign off-chain EIP-3009 authorization (no gas needed)
        sig = encode_payment_signature_header(
            payment_client.create_payment_payload(pr))
        # 5. Retry with signed payment header
        r = session.post(ENDPOINT, json=payload,
                         headers={**headers, X_PAYMENT_HEADER: sig}, timeout=timeout)

    # 6. Parse SSE response — prefer structuredContent
    for line in r.text.splitlines():
        if line.startswith("data:"):
            obj = json.loads(line[5:])
            if "result" in obj:
                sc = obj["result"].get("structuredContent", {})
                if sc: return sc
                for ct in obj["result"].get("content", []):
                    if ct.get("type") == "text":
                        try: return json.loads(ct["text"])
                        except: return {"text": ct["text"]}
    return {}

# ── Examples ──

# Generate text (~$0.01 USDC)
result = call_tool("create_text", {
    "prompt": "Write a haiku about Base blockchain.",
    "provider": "vertex"})
print(result.get("message") or result.get("text"))

# Generate image (~$0.04 USDC)
result = call_tool("create_image", {
    "prompt": "A futuristic city at sunset",
    "provider": "minimax"})
print(f"Resource: {result.get('resource_id')}")
print(f"Wallet link: {result.get('wallet_link')}")

# Edit an image (~$0.01 USDC)
result = call_tool("edit_video", {
    "resource_id": result["resource_id"],
    "action": "add_text",
    "text": "Hello x402",
    "position": "bottom"})
print(f"Edited: {result.get('resource_id')}")

# Analyze media (~$0.01 USDC)
result = call_tool("analyze_media", {
    "resource_id": result["resource_id"],
    "analysis_type": "technical"})
print(f"Type: {result.get('metadata', {}).get('mime_type')}")