Metadata-Version: 2.4
Name: epanya
Version: 0.1.0
Summary: Python SDK for AI agents to discover and purchase products on the Epanya marketplace
Project-URL: Homepage, https://docs.epanya.ai/sdk-usage.html
Project-URL: Repository, https://github.com/Ben1010Tennyson/epanya
Project-URL: Documentation, https://docs.epanya.ai
License: MIT
Requires-Python: >=3.10
Provides-Extra: dev
Requires-Dist: httpx>=0.27; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.23; extra == 'dev'
Requires-Dist: pytest>=8.0; extra == 'dev'
Description-Content-Type: text/markdown

# epanya

Python SDK for AI agents to discover and purchase products on the [Epanya](https://epanya.ai) autonomous commerce marketplace.

Payments are settled in USDC via the [x402 protocol](https://www.x402.org/) on Base L2.

---

## Installation

```bash
pip install epanya
```

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

---

## Quick start (3 lines)

```python
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]`.

```python
# 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.

```python
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.

```python
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.

```python
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.

```python
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.

```python
from typing import Any

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

### Test signer (development only)

```python
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:

```python
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

```python
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

```python
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

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

---

## License

MIT
