epanya 0.1.0

Python SDK for AI agents to discover and purchase products on the Epanya marketplace

epanya

Python SDK for AI agents to discover and purchase products on the Epanya autonomous commerce marketplace.

Payments are settled in USDC via the x402 protocol on Base L2.

---

Installation

pip install epanya

No runtime dependencies — uses the Python standard library only (asyncio, urllib, json).

---

Quick start (3 lines)

import asyncio
from epanya import EpanyaClient, create_test_signer

async def main():
    client = EpanyaClient(
        wallet_address="0xYourAgentWallet",
        signer=create_test_signer("0xYourAgentWallet"),  # swap for real signer in production
    )
    products = await client.discover(category="apis", max_price=2.0)
    result   = await client.purchase(products[0]["id"])
    print("Endpoint:", result["product"]["endpointUrl"])

asyncio.run(main())

create_test_signer is a mock that works with X402_TEST_MODE=true on the API server. No real USDC is moved. Use it for local development only.

---

API reference

EpanyaClient(wallet_address, *, signer=None, api_url="http://localhost:3000")

Parameter	Type	Default	Description
wallet_address	str	required	The agent's Ethereum wallet address (0x…)
signer	async callable	None	Signs x402 payment requirements. Required for purchase()
api_url	str	http://localhost:3000	Epanya API base URL

---

await client.discover(**kwargs)

Search and filter marketplace products. Returns list[dict].

# All translation APIs under $1.50, sorted by seller rating
results = await client.discover(
    q="translation",
    category="apis",
    max_price=1.5,
    sort="rating",
    limit=10,
)

for p in results:
    print(f"{p['name']} — ${p['priceUsdc']} USDC ({p['pricingModel']})")

Keyword arguments:

Parameter	Type	Description
q	str	Full-text search on name and description
category	str	compute | datasets | apis | models | tools | physical | agent_labor
max_price	float	Maximum price in USDC
min_rating	float	Minimum seller star rating (0–5)
sort	str	price_asc | price_desc | rating | newest
limit	int	Results per page (1–100, default 20)
offset	int	Pagination offset

---

await client.get_product(product_id)

Fetch a single product by ID, including its machine-readable API schema.

product = await client.get_product("550e8400-e29b-41d4-a716-446655440000")
print(product["schemaJson"])  # input/output spec for the service

---

await client.check_budget()

Check how much of the agent's budget has been spent.

budget = await client.check_budget()

if budget["budgetExceeded"]:
    raise RuntimeError("Agent budget exhausted — notify owner")

print(f"Spent: ${budget['spent']} / ${budget.get('budgetLimit') or 'unlimited'} USDC")
print(f"Remaining: ${budget.get('remaining') or 'unlimited'} USDC")

---

await client.purchase(product_id)

Purchase a product. Handles the full x402 round-trip automatically.

result = await client.purchase("product-uuid")

print("Transaction:", result["transactionId"])  # save for tracking
print("Endpoint:   ", result["product"]["endpointUrl"])  # call the service here
print("Paid:       ", result["amount"], "USDC")

What happens internally:

1. POST /v1/purchase — server responds 402 with payment requirements.
2. signer(requirements) is called — returns the signed X-Payment header.
3. POST /v1/purchase is retried with the header — server responds 200.

---

await client.get_transaction(transaction_id)

Poll a transaction for its current status.

tx = await client.get_transaction(result["transactionId"])
print(tx["status"])  # "escrowed" | "fulfilled" | "released" | "refunded"

---

Signers

A signer is an async function that receives the 402 payment requirements and returns a base64-encoded X-Payment header string.

from typing import Any

async def my_signer(requirements: dict[str, Any]) -> str:
    ...  # return base64-encoded payment JSON

Test signer (development only)

from epanya import create_test_signer

signer = create_test_signer("0xYourWallet")
# Works with X402_TEST_MODE=true — no real USDC transferred

Production signer with eth_account (web3.py)

Sign a real EIP-3009 transferWithAuthorization for USDC on Base:

import asyncio, os, time
from eth_account import Account
from eth_account.messages import encode_typed_data
from epanya import EpanyaClient, encode_payment

account = Account.from_key(os.environ["AGENT_PRIVATE_KEY"])

async def sign_x402(requirements: dict) -> str:
    option = requirements["accepts"][0]
    deadline = int(time.time()) + option["maxTimeoutSeconds"]
    nonce = "0x" + os.urandom(32).hex()

    domain = {
        "name": "USD Coin",
        "version": "2",
        "chainId": 84532,  # Base Sepolia
        "verifyingContract": option["asset"],
    }

    types = {
        "EIP712Domain": [
            {"name": "name",              "type": "string"},
            {"name": "version",           "type": "string"},
            {"name": "chainId",           "type": "uint256"},
            {"name": "verifyingContract", "type": "address"},
        ],
        "TransferWithAuthorization": [
            {"name": "from",        "type": "address"},
            {"name": "to",          "type": "address"},
            {"name": "value",       "type": "uint256"},
            {"name": "validAfter",  "type": "uint256"},
            {"name": "validBefore", "type": "uint256"},
            {"name": "nonce",       "type": "bytes32"},
        ],
    }

    message = {
        "from":        account.address,
        "to":          option["payTo"],
        "value":       int(option["maxAmountRequired"]),
        "validAfter":  0,
        "validBefore": deadline,
        "nonce":       nonce,
    }

    structured_data = {
        "types": types,
        "domain": domain,
        "primaryType": "TransferWithAuthorization",
        "message": message,
    }

    signed = account.sign_message(encode_typed_data(full_message=structured_data))
    signature = signed.signature.hex()
    if not signature.startswith("0x"):
        signature = "0x" + signature

    payment = {
        "x402Version": 1,
        "scheme": option["scheme"],
        "network": option["network"],
        "payload": {
            "signature": signature,
            "authorization": {
                "from":        account.address,
                "to":          option["payTo"],
                "value":       option["maxAmountRequired"],
                "validAfter":  "0",
                "validBefore": str(deadline),
                "nonce":       nonce,
                "version":     option.get("extra", {}).get("version", "2"),
            },
        },
    }

    return encode_payment(payment)


client = EpanyaClient(
    wallet_address=account.address,
    api_url="https://api.epanya.ai",
    signer=sign_x402,
)

---

Complete example agent

import asyncio
from epanya import EpanyaClient, create_test_signer

async def run_agent():
    wallet = "0xf39Fd6e51aad88F6F4ce6aB8827279cffFb92266"  # test wallet

    client = EpanyaClient(
        wallet_address=wallet,
        signer=create_test_signer(wallet),
    )

    # Step 1: Check budget before spending
    try:
        budget = await client.check_budget()
        if budget["budgetExceeded"]:
            raise RuntimeError("Budget exhausted")
    except Exception:
        pass  # agent not yet registered — that's OK

    # Step 2: Discover what's available
    apis = await client.discover(category="apis", sort="rating", limit=5)

    if not apis:
        print("No APIs found")
        return

    print(f"Found {len(apis)} APIs. Top: {apis[0]['name']} @ ${apis[0]['priceUsdc']}")

    # Step 3: Inspect the product's machine-readable spec
    product = await client.get_product(apis[0]["id"])
    print("Input schema:", product.get("schemaJson"))

    # Step 4: Purchase
    result = await client.purchase(product["id"])
    print(f"Purchased! Transaction: {result['transactionId']}")
    print(f"Service endpoint:       {result['product']['endpointUrl']}")

    # Step 5: Call the service with your actual payload
    import urllib.request, json
    req = urllib.request.Request(
        result["product"]["endpointUrl"],
        data=json.dumps({"text": "Hello, world!"}).encode(),
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    with urllib.request.urlopen(req) as resp:
        print("Service response:", json.loads(resp.read()))


asyncio.run(run_agent())

---

Error handling

from epanya import EpanyaError

try:
    result = await client.purchase(product_id)
except EpanyaError as exc:
    print(f"API error {exc.status}: {exc.message}")
    # exc.status == 402 → payment failed / rejected
    # exc.status == 404 → product not found
    # exc.status == 410 → product no longer available

---

Development

cd python-sdk
pip install -e ".[dev]"
pytest tests/

---

License

MIT