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:
pending→shippedshipped→in_transit,exceptionin_transit→out_for_delivery,delivered,exceptionout_for_delivery→delivered,exceptionexception→in_transit,out_for_delivery,delivereddelivered→ (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-Keyheaders for propose/accept/reject automatically; include a stableclient_proposal_idto dedupe across process restarts. - Listing uses your SDK
agent_idwith an optionalrolefilter (buyerorseller).
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
aftercursor with the last seenevent_idto 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
- Buyer creates agreement → Links to merchant's service via
service_id - Merchant accepts → Escrow funded with prepaid usage
- Buyer calls proxy →
sdk.proxy_request(agreement_id, path, ...) - Paegents injects auth → Fetches merchant's API key and adds to request
- Request forwarded → Sent to merchant's API with authentication
- Usage tracked → Counter incremented automatically
- 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)→A2APaymentResponsecheck_a2a_status(txn_id)→A2AStatusResponse
Service Catalog Methods (Gap 2)
register_service(service)→ServiceDetailsupdate_service(service_id, **kwargs)→Dict[str, Any]delete_service(service_id)→Dict[str, Any]get_service(service_id)→ServiceDetailslist_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)→UsageAgreementget_usage_agreement(agreement_id)→UsageAgreementlist_usage_agreements(status=None, limit=100)→List[UsageAgreement]accept_usage_agreement(agreement_id)→UsageAgreementrecord_usage(agreement_id, request)→UsageAgreementcancel_usage_agreement(agreement_id)→Dict[str, Any]dispute_usage_agreement(agreement_id, reason, evidence=None)→UsageAgreement
Legacy MCP Methods
create_payment(request)→PaymentResponsecreate_paypal_payment(request)→PaymentResponsesearch_recipients(query, payment_method="all")→RecipientSearchResultdiscover_paypal_email(email)→EmailDiscoveryResultcheck_balance()→SpendingLimitsverify_receipt(receipt_id)→Dict[str, Any]
Support
- Documentation: docs.paegents.com
- API Reference: docs.paegents.com/api
- Support: support@paegents.com
- Issues: github.com/paegents/python-sdk/issues
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
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Filter files by name, interpreter, ABI, and platform.
If you're not sure about the file name format, learn more about wheel file names.
Copy a direct link to the current filters
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
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
c689b1c16f43f6ff3c29ab374f3a51eb8f786e9017fe36fcad7312b3f6441874
|
|
| MD5 |
0f92587769270dc7231f04080a6b12f4
|
|
| BLAKE2b-256 |
f6d395ca7738bebd69f54119fc044b3da7625e7849d71c7220741b9f8df67172
|
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
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
218a9fabc87ef601d1c49f0692f599c3daf67a719401b6e7d6a0b054f91a866f
|
|
| MD5 |
87cde4c1e90b8a61cec3bacf544f51ac
|
|
| BLAKE2b-256 |
98566766d7c5a48f85fc9bd8fd9abd1b5499f91b007263694bec992aa831d51a
|