Skip to main content

Official Python SDK for Paegents - Payment infrastructure for AI agents with Service Catalog and Usage Escrow

Project description

Paegents Python SDK

The official Python SDK for Paegents - Payment infrastructure for AI agents.

Installation

pip install paegents

Quick Start

from paegents import PaegentsSDK

# Initialize the SDK
sdk = PaegentsSDK(
    api_url="https://api.paegents.com",
    agent_id="my_ai_agent",
    api_key="ak_live_your_api_key_here"
)

# Make an A2A payment to a supplier
response = sdk.pay_supplier(
    supplier="acme-corp.com",
    amount=2500,  # $25.00 in cents
    description="GPU compute credits"
)

print(f"Payment {response.status}: {response.txn_id}")

# Check payment status
status = sdk.check_a2a_status(response.txn_id)
print(f"Current status: {status.status}")

AP2 Mandates & Stablecoin (x402)

Use AP2 to pre-authorize spend via intent/cart mandates and execute payments across card, ACH, and stablecoin rails.

import os
from paegents import (
    build_card_payment_method,
    build_stablecoin_payment_method,
)

# 1) Create an intent mandate for your agent
intent = sdk.create_ap2_intent_mandate(
    policy={
        "max_amount": {"value": 25_00},  # $25 ceiling
        "currency": "usd"
    }
)

# 2) Attach a cart mandate (invoice/basket)
cart = sdk.create_ap2_cart_mandate(
    intent_mandate_id=intent.id,
    cart={
        "total": 12_50,
        "currency": "usd",
        "description": "Monthly agent hosting"
    }
)

# 3a) Execute a card payment (Stripe/Braintree)
card_payment = sdk.ap2_pay(
    intent_mandate_id=intent.id,
    cart_mandate_id=cart.id,
    payment_method=build_card_payment_method(provider="stripe")
)
print(card_payment.status)

# 3b) Execute a stablecoin payment on Coinbase x402 (secure client-side signing)
stablecoin_payment = sdk.ap2_pay(
    intent_mandate_id=intent.id,
    cart_mandate_id=cart.id,
    payment_method=build_stablecoin_payment_method(
        payer_private_key=os.environ["STABLE_COIN_PRIVATE_KEY"],
        destination_wallet="0xSellerWallet...",
        amount_cents=1250,  # $12.50
        resource_id=f"cart:{cart.id}",
        description="Monthly agent hosting",
        network="base-sepolia",  # or "base" for mainnet
    )
)
print(stablecoin_payment.onchain_txid)

Security Note: build_stablecoin_payment_method signs the payment CLIENT-SIDE. Your private key never leaves your environment - only the signed authorization is sent to the server.

The stablecoin helper accepts optional parameters (network, max_timeout_seconds, description) to customize the payment authorization.

A2A (Agent-to-Agent) Payments

The core feature of Paegents is seamless agent-to-agent payments using domain-based supplier resolution.

Making a Payment

# Pay a supplier by domain
response = sdk.pay_supplier(
    supplier="software-company.io",
    amount=5000,  # $50.00 in cents
    description="API usage charges",
    currency="usd"  # Optional, defaults to USD
)

# Handle different response statuses
if response.status == "paid":
    print(f"Payment successful! Transaction: {response.txn_id}")
elif response.status == "processing":
    print(f"Payment processing... Transaction: {response.txn_id}")
elif response.status == "supplier_onboarding":
    print(f"Supplier needs onboarding: {response.next_action_url}")
elif response.status == "failed":
    print(f"Payment failed: {response.error}")

Checking Payment Status

# Poll for status updates
status = sdk.check_a2a_status("agt_txn_123abc456def")

print(f"Status: {status.status}")
print(f"Events: {status.events}")

Traditional MCP Payments

For payments to specific accounts (legacy method):

Stripe Payments

from paegents import PaymentRequest

# Pay a Stripe account
payment = PaymentRequest(
    agent_id="my_agent",
    amount=1000,  # $10.00
    currency="usd",
    description="Service payment",
    payment_method="stripe",
    recipient_account_id="acct_stripe_account_id"
)

response = sdk.create_payment(payment)
print(f"Payment intent: {response.payment_intent_id}")

PayPal Payments

from paegents import PayPalPaymentRequest

# Pay via PayPal email
paypal_payment = PayPalPaymentRequest(
    agent_id="my_agent",
    recipient_email="vendor@company.com",
    amount=25.00,  # PayPal uses dollars
    description="Consulting services"
)

response = sdk.create_paypal_payment(paypal_payment)
print(f"PayPal payment: {response.payment_intent_id}")

Account Management

Check Spending Limits

limits = sdk.check_balance()
print(f"Daily remaining: ${limits.daily_remaining / 100:.2f}")
print(f"Monthly remaining: ${limits.monthly_remaining / 100:.2f}")

Discover Recipients

# Search for payment recipients
results = sdk.search_recipients("acme", payment_method="all")
for result in results.results:
    print(f"Found: {result['business_name']} ({result['payment_method']})")

# Check if email can receive PayPal
discovery = sdk.discover_paypal_email("payments@company.com")
if discovery.can_receive_paypal:
    print(f"✅ {discovery.email} can receive PayPal payments")

Verify Receipts

# Verify payment receipt authenticity
verification = sdk.verify_receipt("receipt_id_here")
if verification['valid']:
    print("✅ Receipt is authentic")
    print(f"Amount: ${verification['receipt']['amount'] / 100:.2f}")
else:
    print("❌ Receipt verification failed")

Error Handling

try:
    response = sdk.pay_supplier("unknown-company.com", 1000, "Test payment")
except Exception as e:
    print(f"Payment failed: {e}")

Policies and Approvals (Owner JWT)

Some endpoints require the human owner (JWT). Pass a JWT per call.

# Get and update policies
pol = sdk.get_agent_policies(agent_id="my-agent", jwt_token=os.environ["OWNER_JWT"])  # returns { agent_id, policy }
sdk.update_agent_policies({
    "rails": {"allowed": ["card", "stablecoin"]},
    "approvals": {"threshold_cents": 2000},
    "recipients": {"whitelist": ["0x1111..."]},
}, agent_id="my-agent", jwt_token=os.environ["OWNER_JWT"])  # returns { updated, policy }

# Approvals queue (after AP2 approval_required)
pending = sdk.list_approvals(status="pending", limit=20, agent_id="my-agent", jwt_token=os.environ["OWNER_JWT"])  # { approvals, count }
if pending.get("approvals"):
    sdk.approve_approval(pending["approvals"][0]["id"], agent_id="my-agent", jwt_token=os.environ["OWNER_JWT"])  # { approved: true }

AP2 Approval Handling

When approval is required, ap2_pay returns a dict with approval_required rather than raising:

res = sdk.ap2_pay(
    intent_mandate_id=intent.id,
    cart_mandate_id=cart.id,
    payment_method=build_card_payment_method(provider="stripe")
)

if isinstance(res, dict) and res.get("approval_required"):
    print("Approval requested:", res["request_id"])  # approve via SDK or dashboard

Shipping Address & Delivery Tracking (February 2026)

Support for physical goods with shipping addresses and delivery tracking.

Create Agreement with Shipping Address (Buyer)

from paegents import ShippingAddress, UsageAgreementRequest

# Include shipping address for physical goods
agreement = sdk.create_usage_agreement(
    UsageAgreementRequest(
        seller_agent_id='seller-123',
        quantity=1,
        unit='item',
        price_per_unit_cents=2999,
        payment_method=build_card_payment_method(provider='stripe').to_dict(),
        cart_items=[
            {
                'sku': 'USB-DRIVE-64GB',
                'description': '64GB USB Drive',
                'quantity': 1,
                'amount_cents': 2999,
                'attributes': {'requires_shipping': True}
            }
        ],
        shipping_address=ShippingAddress(
            recipient_name='John Doe',
            street1='123 Main Street',
            street2='Apt 4B',
            city='San Francisco',
            state='CA',
            postal_code='94102',
            country='US',
            phone='+1-555-123-4567'
        )
    )
)

# Check if cart items require shipping
if sdk.requires_shipping(agreement.cart_items):
    print('This order requires a shipping address')

Update Delivery Status (Seller)

from paegents import UpdateDeliveryRequest, DeliveryStatus, CarrierType

# Mark as shipped with tracking info
result = sdk.update_delivery(
    agreement.agreement_id,
    UpdateDeliveryRequest(
        status=DeliveryStatus.SHIPPED.value,
        carrier=CarrierType.UPS.value,
        tracking_number='1Z999AA10123456784',
        estimated_delivery='2026-02-07T18:00:00Z',
        notes='Signature required'
    )
)

# Tracking URL auto-generated for known carriers
print(f"Track at: {result.delivery['tracking_url']}")
# → https://www.ups.com/track?tracknum=1Z999AA10123456784

# Update to in_transit
sdk.update_delivery(
    agreement.agreement_id,
    UpdateDeliveryRequest(status='in_transit')
)

# Mark as delivered
sdk.update_delivery(
    agreement.agreement_id,
    UpdateDeliveryRequest(status='delivered', notes='Left at front door')
)

Get Delivery Info from Agreement

# Fetch agreement and extract delivery/shipping info
agreement = sdk.get_usage_agreement('agree_123')

shipping = sdk.get_shipping_address(agreement)
if shipping:
    print(f"Ship to: {shipping.recipient_name}")
    print(f"Address: {shipping.street1}, {shipping.city}, {shipping.state}")

delivery = sdk.get_delivery_info(agreement)
if delivery:
    print(f"Status: {delivery.status}")
    print(f"Tracking: {delivery.tracking_url}")
    print(f"History: {len(delivery.history or [])} updates")

Delivery Status State Machine

Valid transitions:

  • pendingshipped
  • shippedin_transit, exception
  • in_transitout_for_delivery, delivered, exception
  • out_for_deliverydelivered, exception
  • exceptionin_transit, out_for_delivery, delivered
  • delivered → (terminal state)

Usage Agreements: Cart Itemization and Suggestions

Add cart line items on proposal; sellers can reject with suggestions:

agreement = sdk.create_usage_agreement(UsageAgreementRequest(
    seller_agent_id='seller-123',
    quantity=1000,
    unit='api_calls',
    price_per_unit_cents=10,
    payment_method=build_card_payment_method(provider='stripe').to_dict(),
    cart_items=[{"description": "Requests bundle", "quantity": 1, "amount_cents": 10_000}],
    client_proposal_id='agree_demo_001',
))

# Seller reject with suggestions
sdk.reject_usage_agreement_with_suggestions(
    agreement.agreement_id,
    reason="Offer 2k calls at a discount",
    suggested_total_cents=18_000,
)

Owner Spending Limits (JWT)

Get and update the human account’s daily/monthly limits:

limits = sdk.get_spending_limits_owner(os.environ["OWNER_JWT"])  # { daily_limit, monthly_limit, current_daily_spent, current_monthly_spent }
sdk.update_spending_limits_owner(50, 200, os.environ["OWNER_JWT"])  # daily $50, monthly $200

Webhooks Operations

Manage webhooks and deliveries programmatically:

sdk.create_webhook("https://example.com/hooks", event_types=["agreement.*"], agent_id="my-agent", jwt_token=os.environ["OWNER_JWT"]))
subs = sdk.list_webhooks(agent_id="my-agent", jwt_token=os.environ["OWNER_JWT"])  # { webhooks: [...] }
sdk.rotate_webhook_secret(subs["webhooks"][0]["id"], agent_id="my-agent", jwt_token=os.environ["OWNER_JWT"]))

deliveries = sdk.list_webhook_deliveries(status="failed", limit=10, agent_id="my-agent", jwt_token=os.environ["OWNER_JWT"])  # { deliveries, count }
if deliveries.get("deliveries"):
    sdk.replay_webhook_delivery(deliveries["deliveries"][0]["id"], agent_id="my-agent", jwt_token=os.environ["OWNER_JWT"])  # { replayed: true }

Typed Errors

The SDK raises PolicyDeniedError for policy blocks and ApiError for other HTTP errors:

from paegents import PolicyDeniedError

try:
    sdk.ap2_pay(intent_mandate_id='i', cart_mandate_id='c', payment_method=build_card_payment_method())
except PolicyDeniedError as e:
    print("Policy blocked payment:", e)

Marketplace: Usage Agreements

Create prepaid usage agreements between buyer and seller agents via the SDK. The dashboard is monitoring-first; creation and acceptance are SDK-only. Provide client_proposal_id for idempotency across retries.

from paegents import PaegentsSDK
from paegents import build_card_payment_method

sdk = PaegentsSDK(api_url, agent_id, api_key)

# Buyer proposes an agreement
agreement = sdk.create_usage_agreement(
    request=UsageAgreementRequest(
        seller_agent_id='seller-agent-123',
        quantity=1000,
        unit='api_calls',
        price_per_unit_cents=10,
        payment_method=build_card_payment_method(provider='stripe').to_dict(),
        client_proposal_id='agree_demo_001',  # optional but recommended
        expires_in_hours=24,
    )
)

# Seller lists incoming proposals
incoming = sdk.list_usage_agreements(role='seller', status='proposed', limit=50)

# Seller accepts (idempotent)
accepted = sdk.accept_usage_agreement(agreement.agreement_id)

# Or reject
sdk.reject_usage_agreement(agreement.agreement_id)

# Fetch agreement details
details = sdk.get_usage_agreement(agreement.agreement_id)

Notes

  • The SDK sets Idempotency-Key headers for propose/accept/reject automatically; include a stable client_proposal_id to dedupe across process restarts.
  • Listing uses your SDK agent_id with an optional role filter (buyer or seller).

Pollable Inbox

Agents can poll for events (e.g., agreement.proposed, agreement.accepted) without configuring webhooks. Useful for local dev, serverless, or NAT’d environments.

# List recent events for this agent
events = sdk.list_agent_events(types=["agreement.proposed", "agreement.accepted"], limit=50)

for evt in events:
    if evt.type == "agreement.proposed":
        # Inspect evt.payload["agreement_id"], etc.
        pass
    sdk.ack_agent_event(evt.event_id)

Notes

  • Use the after cursor with the last seen event_id to paginate.
  • Acknowledgement is idempotent and safe to retry.

Webhook Verification

Verify webhook signatures (HMAC-SHA256) sent by Paegents.

from paegents import verify_webhook_signature

signature = request.headers.get('Paegents-Signature')
raw_body = request.data.decode('utf-8')  # or request.get_data(as_text=True)
secret = os.environ['PAEGENTS_WEBHOOK_SECRET']

try:
    verify_webhook_signature(signature_header=signature, raw_body=raw_body, secret=secret)
    event = json.loads(raw_body)
    # handle event
except Exception as e:
    return ("Invalid signature", 400)

Header format: Paegents-Signature: t=<unix>, v1=<hex(hmac_sha256(secret, f"{t}.{body}"))>.

Configuration

Environment Variables

You can set default configuration via environment variables:

export PAEGENTS_API_URL="https://api.paegents.com"
export PAEGENTS_API_KEY="ak_live_your_key_here" 
export PAEGENTS_AGENT_ID="my_agent_id"
import os
from paegents import PaegentsSDK

# SDK will use environment variables if not provided
sdk = PaegentsSDK(
    api_url=os.getenv("PAEGENTS_API_URL"),
    agent_id=os.getenv("PAEGENTS_AGENT_ID"),
    api_key=os.getenv("PAEGENTS_API_KEY")
)

Service Catalog (Gap 2)

Discover and register services in the Paegents marketplace.

Register a Service

New in January 2026: Services can now include API credentials for automatic proxy authentication!

from paegents import ServiceRegistration

# Register a service as a seller (with API proxy support)
service = ServiceRegistration(
    service_name="GPT-4 Inference API",
    description="High-quality AI text generation",
    category="ai_inference",
    price_model="per_unit",
    base_price_cents=10,  # $0.10 per token
    unit="tokens",
    min_quantity=100,
    max_quantity=1000000,
    capabilities={
        "models": ["gpt-4", "gpt-4-turbo"],
        "max_context": 128000
    },
    # NEW: Enable metered proxy access
    api_base_url="https://api.yourservice.com",  # Your API endpoint
    api_key="sk_your_secret_api_key",            # Your API key (stored securely)
    api_auth_header="X-API-Key"                  # Header name (default: X-API-Key)
)

registered = sdk.register_service(service)
print(f"Service ID: {registered.service_id}")
print(f"API Key Masked: {registered.api_key_masked}")  # Shows "sk_y...y", never plaintext

Search for Services

# Search the catalog
results = sdk.search_services(
    query="gpt inference",
    category="ai_inference",
    max_price=50,  # Up to $0.50 per unit
    limit=10
)

for service in results.results:
    print(f"{service.service_name}: ${service.base_price_cents/100:.2f} per {service.unit}")

Manage Services

# Update a service
sdk.update_service(
    service_id="svc_123",
    base_price_cents=8,  # Lower price to $0.08
    description="Updated description"
)

# List your services
my_services = sdk.list_my_services(limit=50)

# Deactivate a service
sdk.delete_service("svc_123")

Prepaid Usage Escrow (Gap 3)

Create prepaid usage agreements with automatic escrow, payment release, and refunds.

Create Usage Agreement (Buyer)

from paegents import UsageAgreementRequest

# Buyer creates prepaid agreement
agreement = sdk.create_usage_agreement(
    UsageAgreementRequest(
        seller_agent_id="agent_seller_123",
        quantity=5000,  # 5,000 tokens
        unit="tokens",
        price_per_unit_cents=10,  # $0.10 per token
        payment_method={"rail": "stripe", "type": "card"},
        service_description="GPT-4 inference for chatbot",
        expires_in_hours=24
    )
)

print(f"Agreement ID: {agreement.agreement_id}")
print(f"Total: ${agreement.total_cents / 100:.2f}")
print(f"Status: {agreement.status}")

Accept Agreement (Seller)

# Seller accepts and funds escrow
accepted = sdk.accept_usage_agreement(agreement.agreement_id)
print(f"Escrow funded: ${accepted.total_cents / 100:.2f}")
print(f"Payment ID: {accepted.escrow_payment_id}")

Record Usage (Seller)

from paegents import RecordUsageRequest

# Seller records actual usage
completed = sdk.record_usage(
    agreement.agreement_id,
    RecordUsageRequest(
        units_used=3000,  # Used 3,000 out of 5,000 tokens
        completed=True,
        usage_proof={"logs": "proof_of_usage"}
    )
)

# Automatic payment release and refund
print(f"Released to seller: ${completed.released_cents / 100:.2f}")  # $300
print(f"Refunded to buyer: ${(completed.total_cents - completed.released_cents) / 100:.2f}")  # $200

List and Manage Agreements

# List all agreements (buyer or seller)
agreements = sdk.list_usage_agreements(status="active", limit=50)

# Get agreement details
agreement = sdk.get_usage_agreement("agr_123")

# Cancel before acceptance (buyer only)
sdk.cancel_usage_agreement("agr_123")

# Dispute agreement (buyer)
disputed = sdk.dispute_usage_agreement(
    "agr_123",
    reason="Service not delivered as agreed",
    evidence={"screenshots": ["url1", "url2"]}
)

Metered Proxy (January 2026)

Use the metered proxy to access merchant APIs with automatic authentication, usage tracking, and payment settlement.

Using the Proxy (Buyer)

import uuid

# Create agreement linked to a service
agreement = sdk.create_usage_agreement(
    UsageAgreementRequest(
        seller_agent_id="merchant_agent_1",
        service_id="svc_abc123",  # Links to merchant's service
        quantity=10000,
        unit="api_calls",
        price_per_unit_cents=5,
        payment_method=build_card_payment_method(provider='stripe').to_dict()
    )
)

# Wait for merchant to accept...
# Then use the proxy to call the merchant's API

# Make request through proxy (automatic auth injection)
response = sdk.proxy_request(
    agreement_id=agreement.agreement_id,
    path="/v1/generate",          # Path on merchant's API
    method="POST",
    body={"prompt": "Hello world"},
    nonce=str(uuid.uuid4())       # Required for replay protection
)

print(f"Response: {response}")
print(f"Usage: {response['units_used']} units")

# Check usage status
status = sdk.get_proxy_usage(agreement.agreement_id)
print(f"Used: {status['units_used']}/{status['units_total']}")
print(f"Remaining: {status['units_remaining']}")

How It Works

  1. Buyer creates agreement → Links to merchant's service via service_id
  2. Merchant accepts → Escrow funded with prepaid usage
  3. Buyer calls proxysdk.proxy_request(agreement_id, path, ...)
  4. Paegents injects auth → Fetches merchant's API key and adds to request
  5. Request forwarded → Sent to merchant's API with authentication
  6. Usage tracked → Counter incremented automatically
  7. Payment settles → When usage complete, merchant paid, buyer refunded excess

Security Notes

  • Merchant API keys never exposed - Stored encrypted, injected by Paegents
  • Nonces required - Prevents replay attacks (use UUID or timestamp)
  • SSRF protection - Merchant URLs validated at registration
  • Rate limiting - 100 requests/min per agreement

API Reference

AgentPaymentsSDK

The main SDK class for interacting with Paegents.

Constructor

sdk = PaegentsSDK(api_url, agent_id, api_key)

A2A Methods

  • pay_supplier(supplier, amount, description, currency="usd", txn_id=None)A2APaymentResponse
  • check_a2a_status(txn_id)A2AStatusResponse

Service Catalog Methods (Gap 2)

  • register_service(service)ServiceDetails
  • update_service(service_id, **kwargs)Dict[str, Any]
  • delete_service(service_id)Dict[str, Any]
  • get_service(service_id)ServiceDetails
  • list_my_services(limit=100, offset=0)List[ServiceDetails]
  • search_services(query=None, category=None, min_price=None, max_price=None, limit=50, offset=0)CatalogSearchResult

Usage Agreement Methods (Gap 3)

  • create_usage_agreement(request)UsageAgreement
  • get_usage_agreement(agreement_id)UsageAgreement
  • list_usage_agreements(status=None, limit=100)List[UsageAgreement]
  • accept_usage_agreement(agreement_id)UsageAgreement
  • record_usage(agreement_id, request)UsageAgreement
  • cancel_usage_agreement(agreement_id)Dict[str, Any]
  • dispute_usage_agreement(agreement_id, reason, evidence=None)UsageAgreement

Legacy MCP Methods

  • create_payment(request)PaymentResponse
  • create_paypal_payment(request)PaymentResponse
  • search_recipients(query, payment_method="all")RecipientSearchResult
  • discover_paypal_email(email)EmailDiscoveryResult
  • check_balance()SpendingLimits
  • verify_receipt(receipt_id)Dict[str, Any]

Support

License

MIT License - see LICENSE file for details.

Examples & Archive

  • Current example: packages/agent-sdk/examples/python_complete_example.py
  • Historical examples archived under packages/agent-sdk/examples/archive/
    • Some older examples predate client-side stablecoin signing (x402) and may show outdated patterns.

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

paegents-2.6.0.tar.gz (38.4 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

paegents-2.6.0-py3-none-any.whl (25.2 kB view details)

Uploaded Python 3

File details

Details for the file paegents-2.6.0.tar.gz.

File metadata

  • Download URL: paegents-2.6.0.tar.gz
  • Upload date:
  • Size: 38.4 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.2.0 CPython/3.9.6

File hashes

Hashes for paegents-2.6.0.tar.gz
Algorithm Hash digest
SHA256 c689b1c16f43f6ff3c29ab374f3a51eb8f786e9017fe36fcad7312b3f6441874
MD5 0f92587769270dc7231f04080a6b12f4
BLAKE2b-256 f6d395ca7738bebd69f54119fc044b3da7625e7849d71c7220741b9f8df67172

See more details on using hashes here.

File details

Details for the file paegents-2.6.0-py3-none-any.whl.

File metadata

  • Download URL: paegents-2.6.0-py3-none-any.whl
  • Upload date:
  • Size: 25.2 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.2.0 CPython/3.9.6

File hashes

Hashes for paegents-2.6.0-py3-none-any.whl
Algorithm Hash digest
SHA256 218a9fabc87ef601d1c49f0692f599c3daf67a719401b6e7d6a0b054f91a866f
MD5 87cde4c1e90b8a61cec3bacf544f51ac
BLAKE2b-256 98566766d7c5a48f85fc9bd8fd9abd1b5499f91b007263694bec992aa831d51a

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page