How to Integrate AI Agents with Stripe

pip install stripe langchain langchain-openai python-dotenv

export STRIPE_SECRET_KEY="sk_live_your_key"  # Use sk_test_ for development
export STRIPE_WEBHOOK_SECRET="whsec_your_webhook_secret"

import os
import stripe
from langchain.tools import tool
from datetime import datetime, timedelta
from dotenv import load_dotenv

load_dotenv()
stripe.api_key = os.getenv("STRIPE_SECRET_KEY")


@tool
def get_payment_failure_summary(hours_back: int = 24) -> str:
    """Get a summary of payment failures in the last N hours with failure reason breakdown."""
    cutoff = int((datetime.now() - timedelta(hours=hours_back)).timestamp())

    # Fetch recent charges
    charges = stripe.Charge.list(
        created={"gte": cutoff},
        limit=100
    )

    failures = [c for c in charges.auto_paging_iter()
                if c.status == "failed" and c.created >= cutoff]

    if not failures:
        return f"No payment failures in the last {hours_back} hours"

    # Classify by failure reason
    reasons = {}
    for charge in failures:
        reason = charge.failure_code or "unknown"
        reasons[reason] = reasons.get(reason, 0) + 1

    total_failed_amount = sum(c.amount for c in failures) / 100
    summary = [f"Payment Failures (last {hours_back}h):",
               f"Total failures: {len(failures)}",
               f"Total amount at risk: ${total_failed_amount:,.2f}",
               "\nFailure reasons:"]
    for reason, count in sorted(reasons.items(), key=lambda x: -x[1]):
        summary.append(f"  - {reason}: {count} ({count/len(failures)*100:.0f}%)")
    return "\n".join(summary)


@tool
def get_mrr_snapshot() -> str:
    """Get current Monthly Recurring Revenue and subscription statistics."""
    subscriptions = stripe.Subscription.list(status='active', limit=100)

    total_mrr = 0
    plan_breakdown = {}

    for sub in subscriptions.auto_paging_iter():
        for item in sub['items']['data']:
            amount = item['price']['unit_amount'] or 0
            interval = item['price']['recurring']['interval']
            plan_name = item['price'].get('nickname') or item['price']['id']

            # Normalize to monthly
            monthly = amount / 12 if interval == 'year' else amount
            total_mrr += monthly

            plan_breakdown[plan_name] = plan_breakdown.get(plan_name, {'count': 0, 'mrr': 0})
            plan_breakdown[plan_name]['count'] += 1
            plan_breakdown[plan_name]['mrr'] += monthly / 100

    result = [f"MRR Snapshot:",
              f"Total MRR: ${total_mrr/100:,.2f}",
              f"ARR: ${total_mrr/100*12:,.2f}",
              "\nPlan breakdown:"]
    for plan, data in sorted(plan_breakdown.items(), key=lambda x: -x[1]['mrr']):
        result.append(f"  - {plan}: {data['count']} subscriptions, ${data['mrr']:,.2f}/mo")
    return "\n".join(result)


@tool
def get_customer_payment_history(customer_email: str) -> str:
    """Look up payment history for a customer by email address."""
    customers = stripe.Customer.list(email=customer_email, limit=1)
    if not customers.data:
        return f"No Stripe customer found with email: {customer_email}"

    customer = customers.data[0]
    charges = stripe.Charge.list(customer=customer.id, limit=20)

    history = [f"Payment history for {customer_email} (ID: {customer.id}):"]
    for charge in charges.auto_paging_iter():
        date = datetime.fromtimestamp(charge.created).strftime("%Y-%m-%d")
        amount = charge.amount / 100
        status = charge.status
        history.append(f"  {date}: ${amount:.2f} — {status}")

    return "\n".join(history) if len(history) > 1 else "No payment history found"


@tool
def list_recent_disputes() -> str:
    """List open payment disputes/chargebacks for review."""
    disputes = stripe.Dispute.list(limit=20)

    open_disputes = [d for d in disputes.auto_paging_iter() if d.status == 'needs_response']
    if not open_disputes:
        return "No disputes currently need response"

    result = [f"Open disputes requiring response ({len(open_disputes)}):"]
    for dispute in open_disputes:
        due_by = datetime.fromtimestamp(dispute.evidence_details['due_by']).strftime("%Y-%m-%d")
        result.append(f"  - ${dispute.amount/100:.2f} | Reason: {dispute.reason} | Due: {due_by} | ID: {dispute.id}")
    return "\n".join(result)

from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate

llm = ChatOpenAI(model="gpt-4o", temperature=0)
tools = [get_payment_failure_summary, get_mrr_snapshot,
         get_customer_payment_history, list_recent_disputes]

prompt = ChatPromptTemplate.from_messages([
    ("system", """You are a financial analyst with access to Stripe payment data.

Provide clear, actionable financial analysis:
- For payment failures: identify patterns and recommend response priorities
- For MRR questions: provide growth context and plan performance
- For customer inquiries: provide accurate payment history with relevant context
- For disputes: summarize evidence gaps and recommend response strategy

Always present financial data clearly with proper currency formatting."""),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

agent = create_tool_calling_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# Generate daily finance briefing
result = executor.invoke({
    "input": "Generate a daily revenue briefing: check payment failures from the last 24 hours, current MRR, and any open disputes."
})
print(result["output"])

from flask import Flask, request, jsonify
import stripe

app = Flask(__name__)

@app.route('/stripe/webhook', methods=['POST'])
def stripe_webhook():
    payload = request.data
    sig_header = request.headers.get('Stripe-Signature')

    try:
        event = stripe.Webhook.construct_event(
            payload, sig_header, os.getenv("STRIPE_WEBHOOK_SECRET")
        )
    except stripe.error.SignatureVerificationError:
        return jsonify({"error": "Invalid signature"}), 400

    if event.type == 'payment_intent.payment_failed':
        payment_intent = event.data.object
        customer_id = payment_intent.get('customer')

        # Trigger agent to analyze and respond
        result = executor.invoke({
            "input": f"""A payment failed for customer {customer_id}.
Payment amount: ${payment_intent['amount']/100:.2f}
Failure reason: {payment_intent.get('last_payment_error', {}).get('message', 'Unknown')}

Look up the customer's payment history and provide a recommended response strategy."""
        })
        # Log result or notify team via Slack
        print(result["output"])

    return jsonify({"received": True})