Stripe Integration¶

Stripe provides payment processing and billing infrastructure for API monetization.

Configuration¶

Environment Variables:

STRIPE_API_KEY=sk_test_...           # Stripe secret key
STRIPE_PUBLISHABLE_KEY=pk_test_...   # Stripe publishable key
STRIPE_WEBHOOK_SECRET=whsec_...      # Webhook signing secret

Location: /home/agent/projects/aegis-core/aegis/monetization/billing.py

Architecture¶

The Stripe integration is part of the broader monetization module:

aegis/monetization/
├── billing.py          # Stripe integration
├── customers.py        # Customer & API key management
├── products.py         # Product catalog
├── pricing.py          # Pricing engine
├── metering.py         # Usage tracking
├── subscriptions.py    # Subscription management
└── services/           # Billable services
    ├── research_report.py
    └── email_draft.py

Database Schema¶

billing_transactions¶

CREATE TABLE billing_transactions (
    id SERIAL PRIMARY KEY,
    transaction_id UUID UNIQUE NOT NULL DEFAULT gen_random_uuid(),
    customer_id TEXT NOT NULL,
    stripe_invoice_id TEXT,
    stripe_charge_id TEXT,
    amount_cents INTEGER NOT NULL,
    currency TEXT DEFAULT 'USD',
    status TEXT NOT NULL,  -- pending, processing, succeeded, failed, refunded, cancelled
    description TEXT,
    metadata JSONB DEFAULT '{}',
    billing_period TEXT,
    created_at TIMESTAMPTZ DEFAULT NOW(),
    updated_at TIMESTAMPTZ DEFAULT NOW()
);

stripe_customers¶

CREATE TABLE stripe_customers (
    id SERIAL PRIMARY KEY,
    customer_id TEXT UNIQUE NOT NULL,  -- Aegis customer ID
    stripe_customer_id TEXT UNIQUE NOT NULL,
    email TEXT,
    name TEXT,
    metadata JSONB DEFAULT '{}',
    created_at TIMESTAMPTZ DEFAULT NOW(),
    updated_at TIMESTAMPTZ DEFAULT NOW()
);

stripe_webhook_events¶

CREATE TABLE stripe_webhook_events (
    id SERIAL PRIMARY KEY,
    event_id TEXT UNIQUE NOT NULL,
    event_type TEXT NOT NULL,
    payload JSONB NOT NULL,
    processed BOOLEAN DEFAULT false,
    created_at TIMESTAMPTZ DEFAULT NOW()
);

Core Classes¶

StripeCustomerManager¶

Manages Stripe customer records:

from aegis.monetization.billing import StripeCustomerManager

manager = StripeCustomerManager()

# Create or get Stripe customer
stripe_customer = manager.get_or_create_customer(
    customer_id="cust_abc123",
    email="user@example.com",
    name="John Doe"
)
# Returns: {"id": "cus_xxx", "customer_id": "cust_abc123", ...}

BillingService¶

Handles invoicing and payments:

from aegis.monetization.billing import BillingService

service = BillingService()

# Create invoice from usage
invoice = await service.create_invoice_from_usage(
    customer_id="cust_abc123",
    billing_period="2026-01"
)
# Returns: {"invoice_id": "in_xxx", "amount_due": 5000, ...}

# Record transaction
transaction = service.record_transaction(
    customer_id="cust_abc123",
    amount_cents=5000,
    description="API usage for January 2026",
    stripe_invoice_id="in_xxx"
)

Product Catalog¶

Defining Products¶

from aegis.monetization.products import ProductCatalog

catalog = ProductCatalog()

# Register product
catalog.register_product(
    product_id="geoint_report",
    name="Geopolitical Intelligence Report",
    description="Custom research report with analysis",
    pricing_type="outcome",  # one-time charge per deliverable
    base_price_cents=99900,   # $999
    metadata={
        "delivery_time": "24-48 hours",
        "format": "PDF + markdown"
    }
)

# Register API product
catalog.register_product(
    product_id="api_call",
    name="API Call",
    pricing_type="usage",  # per-call pricing
    base_price_cents=10,   # $0.10 per call
)

Pricing Models¶

Usage-Based Pricing¶

from aegis.monetization.pricing import PricingEngine

engine = PricingEngine()

# Calculate cost for API calls
cost = engine.calculate_usage_cost(
    product_id="api_call",
    quantity=1000,  # 1000 API calls
    customer_id="cust_abc123"
)
# Returns: {"amount_cents": 10000, "quantity": 1000, "unit_price": 10}

Tiered Pricing¶

# Define pricing tiers
engine.set_tiered_pricing(
    product_id="api_call",
    tiers=[
        {"up_to": 1000, "price_cents": 10},      # $0.10 per call for 0-1000
        {"up_to": 10000, "price_cents": 8},      # $0.08 per call for 1001-10000
        {"up_to": None, "price_cents": 5}        # $0.05 per call for 10001+
    ]
)

# Calculate with tiers
cost = engine.calculate_usage_cost("api_call", quantity=5000, customer_id="cust_abc123")
# Returns: {"amount_cents": 42000, ...}  # 1000*10 + 4000*8

Volume Discounts¶

# Set volume discount
engine.set_volume_discount(
    customer_id="cust_abc123",
    discount_percent=20  # 20% off for high-volume customer
)

cost = engine.calculate_usage_cost("api_call", quantity=1000, customer_id="cust_abc123")
# Returns: {"amount_cents": 8000, "discount_applied": 2000}  # 20% off

Metering Usage¶

Track API Calls¶

from aegis.monetization.metering import UsageMeter

meter = UsageMeter()

# Record API call
meter.record_usage(
    customer_id="cust_abc123",
    product_id="api_call",
    quantity=1,
    metadata={"endpoint": "/api/intel/brief", "status": 200}
)

# Get usage for billing period
usage = meter.get_usage(
    customer_id="cust_abc123",
    product_id="api_call",
    start_date="2026-01-01",
    end_date="2026-01-31"
)
# Returns: {"total_quantity": 1234, "events": [...]}

Webhook Handling¶

Stripe sends webhook events for payment updates.

Endpoint: /api/stripe/webhook

Supported Events¶

invoice.payment_succeeded - Payment completed
invoice.payment_failed - Payment failed
customer.subscription.created - New subscription
customer.subscription.updated - Subscription changed
customer.subscription.deleted - Subscription cancelled
charge.succeeded - Charge completed
charge.failed - Charge failed

Webhook Handler¶

from aegis.monetization.billing import StripeWebhookHandler

handler = StripeWebhookHandler()

# Process webhook (called by FastAPI route)
result = await handler.process_webhook(
    payload=request_body,
    signature=stripe_signature_header
)

Webhook Verification¶

import stripe

def verify_webhook(payload: bytes, signature: str, secret: str) -> dict:
    """Verify webhook signature and parse event."""
    try:
        event = stripe.Webhook.construct_event(
            payload, signature, secret
        )
        return event
    except stripe.error.SignatureVerificationError:
        raise HTTPException(status_code=401, detail="Invalid signature")

Subscriptions¶

Agent-as-Employee Plans¶

from aegis.monetization.subscriptions import SubscriptionManager

manager = SubscriptionManager()

# Create subscription
subscription = manager.create_subscription(
    customer_id="cust_abc123",
    plan_id="analyst",  # junior, analyst, lead, principal
    billing_interval="monthly"
)

# Get active subscription
sub = manager.get_active_subscription(customer_id="cust_abc123")
# Returns: {"plan_id": "analyst", "price_cents": 199900, ...}

# Cancel subscription
manager.cancel_subscription(customer_id="cust_abc123")

Subscription Plans¶

Plan	Price/Month	Features
Junior	$999	40h/month, basic tasks
Analyst	$1,999	80h/month, research, analysis
Lead	$4,999	160h/month, strategic work
Principal	Custom	Unlimited, dedicated agent

Revenue Tracking¶

Get Revenue Metrics¶

from aegis.monetization.billing import BillingService

service = BillingService()

# Get monthly revenue
revenue = service.get_revenue(
    start_date="2026-01-01",
    end_date="2026-01-31"
)
# Returns: {"total_cents": 150000, "transaction_count": 45, ...}

# Get revenue by product
breakdown = service.get_revenue_by_product(
    start_date="2026-01-01",
    end_date="2026-01-31"
)
# Returns: {"geoint_report": 99900, "api_call": 50100, ...}

Customer Management¶

API Keys¶

from aegis.monetization.customers import APIKeyManager

manager = APIKeyManager()

# Create API key
key = manager.create_api_key(
    customer_id="cust_abc123",
    name="Production API Key",
    scopes=["intel:read", "intel:write"]
)
# Returns: {"key": "ak_xxx", "customer_id": "cust_abc123", ...}

# Validate API key
customer = manager.validate_api_key("ak_xxx")
# Returns: {"customer_id": "cust_abc123", "scopes": [...], ...}

# Revoke API key
manager.revoke_api_key("ak_xxx")

Testing Stripe Integration¶

Test Mode¶

Use Stripe test keys for development:

STRIPE_API_KEY=sk_test_...
STRIPE_PUBLISHABLE_KEY=pk_test_...

Test Cards¶

Card Number	Purpose
4242 4242 4242 4242	Successful payment
4000 0000 0000 0002	Card declined
4000 0000 0000 9995	Insufficient funds

Mock Mode¶

If Stripe package is not installed, operations are mocked:

# Gracefully degrades to mock implementation
if not STRIPE_AVAILABLE:
    logger.warning("stripe_package_not_installed",
                  message="Stripe operations will be mocked")

Best Practices¶

Use Idempotency Keys: Prevent duplicate charges
Handle Webhooks Async: Process in background tasks
Store Webhook Events: Keep audit trail in database
Verify Signatures: Always verify webhook signatures
Test Thoroughly: Use Stripe test mode extensively
Monitor Failed Payments: Alert on payment failures
Graceful Degradation: Handle Stripe downtime

Troubleshooting¶

Payment Failed¶

Check Stripe dashboard for error details:

# Get failed charges
charges = stripe.Charge.list(
    customer=stripe_customer_id,
    limit=10
)
for charge in charges:
    if charge.status == "failed":
        print(f"Failed: {charge.failure_message}")

Webhook Not Received¶

Verify webhook endpoint:

# Test webhook locally with Stripe CLI
stripe listen --forward-to localhost:8080/api/stripe/webhook

# Trigger test event
stripe trigger invoice.payment_succeeded

Customer Not Found¶

Ensure Stripe customer is created:

# Check if customer exists
try:
    customer = stripe.Customer.retrieve(stripe_customer_id)
except stripe.error.InvalidRequestError:
    # Create customer
    customer = manager.get_or_create_customer(...)

Monitoring¶

Key Metrics¶

Monthly Recurring Revenue (MRR)
Customer Lifetime Value (LTV)
Churn Rate
Failed Payment Rate
Average Revenue Per User (ARPU)

Alerts¶

Set up alerts for: - Failed payments (notify customer) - Subscription cancellations - High-value transactions - Webhook processing failures