Skip to main content

​
Installation

Install the required dependencies:
pip
pip install requests fastapi uvicorn
This guide uses the requests library for the client. For async operations, consider using httpx or aiohttp. FastAPI is used for the web framework examples.

​
Setup

Create a new file 0xmeta.py to encapsulate the API client: 
import os
import time
import requests
from typing import Optional, Dict, Any

class ZeroXMetaClient:
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = 'https://facilitator.api.0xmeta.ai/v1'
        self.session = requests.Session()
        self.session.headers.update({
            'Content-Type': 'application/json',
            'X-API-Key': self.api_key
        })

    def _request(self, method: str, endpoint: str, **kwargs) -> Dict[str, Any]:
        url = f"{self.base_url}{endpoint}"

        response = self.session.request(method, url, **kwargs)

        if not response.ok:
            error_data = response.json()
            error_msg = error_data.get('error', {}).get('message', f'HTTP {response.status_code}')
            raise Exception(error_msg)

        return response.json()

    def verify_payment(
        self,
        transaction_hash: str,
        chain: str,
        seller_address: str,
        expected_amount: str,
        expected_token: Optional[str] = None,
        metadata: Optional[Dict[str, Any]] = None,
        webhook_url: Optional[str] = None,
        idempotency_key: Optional[str] = None
    ) -> Dict[str, Any]:
        """Verify a blockchain payment transaction."""
        if not idempotency_key:
            idempotency_key = f"verify_{int(time.time())}"

        headers = {'Idempotency-Key': idempotency_key}

        payload = {
            'transaction_hash': transaction_hash,
            'chain': chain,
            'seller_address': seller_address,
            'expected_amount': expected_amount,
            'expected_token': expected_token,
            'metadata': metadata or {},
            'webhook_url': webhook_url
        }

        return self._request('POST', '/verify', headers=headers, json=payload)

    def settle_payment(
        self,
        verification_id: str,
        destination_address: str,
        amount: Optional[str] = None,
        metadata: Optional[Dict[str, Any]] = None,
        idempotency_key: Optional[str] = None
    ) -> Dict[str, Any]:
        """Initiate settlement of a verified payment."""
        if not idempotency_key:
            idempotency_key = f"settle_{int(time.time())}"

        headers = {'Idempotency-Key': idempotency_key}

        payload = {
            'verification_id': verification_id,
            'destination_address': destination_address,
            'amount': amount,
            'metadata': metadata or {}
        }

        return self._request('POST', '/settle', headers=headers, json=payload)

    def get_verification_status(self, verification_id: str) -> Dict[str, Any]:
        """Get the status of a verification."""
        return self._request('GET', f'/verifications/{verification_id}')

    def get_settlement_status(self, settlement_id: str) -> Dict[str, Any]:
        """Get the status of a settlement."""
        return self._request('GET', f'/settlements/{settlement_id}')

​
Basic Usage

​
Verify a Payment

from 0xmeta import ZeroXMetaClient
import os

client = ZeroXMetaClient(api_key=os.getenv('0XMETA_API_KEY'))

# Verify a payment
try:
    result = client.verify_payment(
        transaction_hash='0x1234567890abcdef1234567890abcdef1234567890abcdef1234567890abcdef',
        chain='base',
        seller_address='0x742d35cc6634c0532925a3b844bc9e7595f0beb0',
        expected_amount='1000000000000000000',  # 1 ETH in wei
        expected_token=None,  # None for native token
        metadata={
            'order_id': 'ORDER-123',
            'customer_id': 'CUST-456'
        },
        webhook_url='https://your-app.com/webhooks/settlement',
        idempotency_key='verify_ORDER-123_1234567890'
    )

    print(f"Verification ID: {result['verification_id']}")
    print(f"Status: {result['status']}")

except Exception as e:
    print(f"Verification failed: {e}")

​
Settle a Payment

def settle_payment(verification_id: str):
    try:
        result = client.settle_payment(
            verification_id=verification_id,
            destination_address='0x742d35cc6634c0532925a3b844bc9e7595f0beb0',
            amount=None,  # None to settle full amount
            metadata={
                'payout_id': 'PAYOUT-789'
            },
            idempotency_key=f'settle_{verification_id}_{int(time.time())}'
        )

        print(f"Settlement ID: {result['settlement_id']}")
        print(f"Status: {result['status']}")

        return result
    except Exception as e:
        print(f"Settlement failed: {e}")
        raise

​
FastAPI Integration

​
Complete FastAPI Route Example

from fastapi import FastAPI, HTTPException, Request, Header
from fastapi.responses import JSONResponse
from pydantic import BaseModel
from typing import Optional, Dict, Any
from 0xmeta import ZeroXMetaClient
import os
import time
import json

app = FastAPI()
client = ZeroXMetaClient(api_key=os.getenv('0XMETA_API_KEY'))

# Request models
class VerifyPaymentRequest(BaseModel):
    transaction_hash: str
    chain: str
    seller_address: str
    expected_amount: str
    expected_token: Optional[str] = None
    order_id: Optional[str] = None

class SettlePaymentRequest(BaseModel):
    verification_id: str
    destination_address: str

@app.post('/api/payments/verify')
async def verify_payment(
    data: VerifyPaymentRequest,
    request: Request
):
    try:
        # Get base URL from request
        base_url = f"{request.url.scheme}://{request.url.netloc}"
        
        verification = client.verify_payment(
            transaction_hash=data.transaction_hash,
            chain=data.chain,
            seller_address=data.seller_address,
            expected_amount=data.expected_amount,
            expected_token=data.expected_token,
            metadata={'order_id': data.order_id} if data.order_id else {},
            webhook_url=f"{base_url}/webhooks/settlement",
            idempotency_key=f"verify_{data.order_id}_{int(time.time())}" if data.order_id else None
        )

        return {
            'success': True,
            'verification_id': verification['verification_id'],
            'status': verification['status']
        }
    except Exception as e:
        raise HTTPException(
            status_code=400,
            detail={
                'success': False,
                'error': str(e)
            }
        )

@app.post('/api/payments/settle')
async def settle_payment(data: SettlePaymentRequest):
    try:
        settlement = client.settle_payment(
            verification_id=data.verification_id,
            destination_address=data.destination_address,
            idempotency_key=f"settle_{data.verification_id}_{int(time.time())}"
        )

        return {
            'success': True,
            'settlement_id': settlement['settlement_id'],
            'status': settlement['status']
        }
    except Exception as e:
        raise HTTPException(
            status_code=400,
            detail={
                'success': False,
                'error': str(e)
            }
        )

@app.post('/webhooks/settlement')
async def webhook_handler(
    request: Request,
    x_signature: Optional[str] = Header(None, alias='X-Signature')
):
    try:
        # Get raw body for signature verification (if needed)
        body = await request.body()
        payload = json.loads(body)

        # Verify webhook signature here (see webhooks guide)
        # if not verify_webhook_signature(body, x_signature, WEBHOOK_SECRET):
        #     raise HTTPException(status_code=401, detail='Invalid signature')

        if payload.get('type') == 'settlement.completed':
            settlement_id = payload['data']['settlement_id']
            verification_id = payload['data']['verification_id']
            status = payload['data']['status']

            # Update your database
            update_settlement_status(settlement_id, status)

            print(f"Settlement {settlement_id} completed")

        return {'received': True}
    except Exception as e:
        print(f"Webhook error: {e}")
        raise HTTPException(status_code=400, detail={'error': str(e)})

if __name__ == '__main__':
    import uvicorn
    uvicorn.run(app, host='0.0.0.0', port=3000)

​
Django Integration

​
Django Views Example

from django.http import JsonResponse
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_http_methods
import json
from 0xmeta import ZeroXMetaClient
import os

client = ZeroXMetaClient(api_key=os.getenv('0XMETA_API_KEY'))

@csrf_exempt
@require_http_methods(["POST"])
def verify_payment(request):
    try:
        data = json.loads(request.body)

        verification = client.verify_payment(
            transaction_hash=data['transaction_hash'],
            chain=data['chain'],
            seller_address=data['seller_address'],
            expected_amount=data['expected_amount'],
            expected_token=data.get('expected_token'),
            metadata={'order_id': data.get('order_id')},
            webhook_url=f"{request.scheme}://{request.get_host()}/webhooks/settlement",
            idempotency_key=f"verify_{data.get('order_id')}_{int(time.time())}"
        )

        return JsonResponse({
            'success': True,
            'verification_id': verification['verification_id'],
            'status': verification['status']
        })
    except Exception as e:
        return JsonResponse({
            'success': False,
            'error': str(e)
        }, status=400)

@csrf_exempt
@require_http_methods(["POST"])
def settle_payment(request):
    try:
        data = json.loads(request.body)

        settlement = client.settle_payment(
            verification_id=data['verification_id'],
            destination_address=data['destination_address'],
            idempotency_key=f"settle_{data['verification_id']}_{int(time.time())}"
        )

        return JsonResponse({
            'success': True,
            'settlement_id': settlement['settlement_id'],
            'status': settlement['status']
        })
    except Exception as e:
        return JsonResponse({
            'success': False,
            'error': str(e)
        }, status=400)

​
Error Handling

class ZeroXMetaError(Exception):
    def __init__(self, message: str, code: str = None, details: dict = None):
        self.message = message
        self.code = code
        self.details = details
        super().__init__(self.message)

# Enhanced request method with error handling
def _request(self, method: str, endpoint: str, **kwargs) -> Dict[str, Any]:
    url = f"{self.base_url}{endpoint}"

    try:
        response = self.session.request(method, url, **kwargs)

        if not response.ok:
            error_data = response.json()
            error_info = error_data.get('error', {})

            raise ZeroXMetaError(
                error_info.get('message', f'HTTP {response.status_code}'),
                error_info.get('code', 'unknown_error'),
                error_info.get('details')
            )

        return response.json()
    except requests.exceptions.RequestException as e:
        raise ZeroXMetaError(
            str(e),
            'network_error',
            {'original_error': str(e)}
        )

# Usage with error handling
try:
    result = client.verify_payment(...)
except ZeroXMetaError as e:
    if e.code == 'verification_failed':
        print(f"Payment verification failed: {e.details}")
    elif e.code == 'invalid_request':
        print(f"Invalid request: {e.details}")
    else:
        print(f"API error: {e.message}")
except Exception as e:
    print(f"Unexpected error: {e}")

​
Polling for Status

import time

def wait_for_settlement(settlement_id: str, max_attempts: int = 60) -> Dict[str, Any]:
    """Poll for settlement status with exponential backoff."""
    initial_delay = 5  # 5 seconds

    for attempt in range(max_attempts):
        settlement = client.get_settlement_status(settlement_id)

        if settlement['status'] in ['settled', 'failed']:
            return settlement

        # Exponential backoff: 5s, 10s, 20s, 40s, ...
        delay = initial_delay * (2 ** min(attempt, 5))
        time.sleep(delay)

    raise TimeoutError('Settlement timeout')

# Usage
settlement = client.settle_payment(...)
completed = wait_for_settlement(settlement['settlement_id'])
print(f"Settlement completed: {completed}")

​
Async Support with httpx

import httpx
import asyncio

class AsyncZeroXMetaClient:
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = 'https://facilitator.api.0xmeta.ai/v1'
        self.client = httpx.AsyncClient(
            headers={
                'Content-Type': 'application/json',
                'X-API-Key': self.api_key
            }
        )

    async def verify_payment(self, **kwargs):
        idempotency_key = kwargs.get('idempotency_key') or f"verify_{int(time.time())}"

        headers = {'Idempotency-Key': idempotency_key}
        payload = {
            'transaction_hash': kwargs['transaction_hash'],
            'chain': kwargs['chain'],
            'seller_address': kwargs['seller_address'],
            'expected_amount': kwargs['expected_amount'],
            'expected_token': kwargs.get('expected_token'),
            'metadata': kwargs.get('metadata', {}),
            'webhook_url': kwargs.get('webhook_url')
        }

        response = await self.client.post(
            f"{self.base_url}/verify",
            headers=headers,
            json=payload
        )
        response.raise_for_status()
        return response.json()

    async def close(self):
        await self.client.aclose()

# Usage
async def main():
    client = AsyncZeroXMetaClient(api_key=os.getenv('0XMETA_API_KEY'))

    try:
        result = await client.verify_payment(
            transaction_hash='0x...',
            chain='base',
            seller_address='0x...',
            expected_amount='1000000000000000000'
        )
        print(result)
    finally:
        await client.close()

asyncio.run(main())

​
Webhook Signature Verification

import hmac
import hashlib

def verify_webhook_signature(payload: bytes, signature: str, secret: str) -> bool:
    """Verify webhook signature using HMAC-SHA256."""
    expected_signature = hmac.new(
        secret.encode('utf-8'),
        payload,
        hashlib.sha256
    ).hexdigest()

    return hmac.compare_digest(expected_signature, signature)

# FastAPI example
import json

@app.post('/webhooks/settlement')
async def webhook_handler(
    request: Request,
    x_signature: Optional[str] = Header(None, alias='X-Signature')
):
    # Get raw body for signature verification
    body = await request.body()
    
    if not verify_webhook_signature(body, x_signature, WEBHOOK_SECRET):
        raise HTTPException(status_code=401, detail='Invalid signature')
    
    # Process webhook...
    data = json.loads(body)
    return {'received': True}

​
Best Practices

# .env
0XMETA_API_KEY=your_api_key_here

# In your code
from dotenv import load_dotenv
import os

load_dotenv()
client = ZeroXMetaClient(api_key=os.getenv('0XMETA_API_KEY'))
# Store idempotency keys in your database idempotency_key =
f"verify_{order_id}_{int(time.time())}" # Check if already processed existing
= db.find_verification(idempotency_key) if existing: return existing result =
client.verify_payment( ..., idempotency_key=idempotency_key )
db.save_verification(idempotency_key, result) ```
</Accordion>

{" "}

<Accordion title="Use Type Hints">
```python from typing import Optional, Dict, Any, Literal def verify_payment(
self, transaction_hash: str, chain: Literal['base', 'base-sepolia'],
seller_address: str, expected_amount: str, expected_token: Optional[str] =
None, metadata: Optional[Dict[str, Any]] = None ) -> Dict[str, Any]: #
Implementation... ```
</Accordion>

<Accordion title="Use Context Managers">
  ```python
  # For async clients
  async with AsyncZeroXMetaClient(api_key=api_key) as client:
      result = await client.verify_payment(...)
  
  # For sync clients with session management
  with requests.Session() as session:
      client = ZeroXMetaClient(api_key=api_key)
      client.session = session
      result = client.verify_payment(...)
For production applications, consider using httpx for async operations or implementing connection pooling with requests.Session().
Always use webhooks instead of polling for settlement status updates. It’s more efficient and provides instant notifications.