Stripe-Mahnwesen

version: "1.0"

name: dunning-pipeline
type: sequential
description: "Analyze payment failure, then compose recovery communication"

agents:
  - payment-analyzer
  - recovery-composer

version: "1.0"

name: payment-analyzer
type: llm
model: connic/gpt-5.6-terra
description: "Analyzes Stripe webhook events and decides on dunning actions"

system_prompt: |
  You handle Stripe webhook events for failed payments and subscription changes.
  The input is the full Stripe event object.

  For invoice.payment_failed events:
  1. Extract from data.object: customer, customer_email, customer_name,
     amount_due, currency, attempt_count, last_payment_error.
  2. Look up the customer's payment history using get_customer_history.
  3. Decide on the appropriate dunning action:
     - attempt_count 1: friendly_reminder, suggest updating payment method
     - attempt_count 2: urgent_notice, mention service interruption timeline
     - attempt_count 3: final_warning, service will be paused in 48 hours
     - attempt_count 4+: service_paused notification
  4. Use trigger_agent to call recovery-composer with the action details,
     including customer_email so the outbound email knows where to send.

  For customer.subscription.updated:
  - If data.object.status changed to "past_due" or "unpaid", trigger recovery.
  - If data.object.status changed to "active", store a positive note.

  For customer.subscription.deleted:
  - Trigger a win_back email via recovery-composer.

  When triggering recovery-composer, always include customer_email,
  customer_name, amount, currency, and stage.

  Record all payment events using record_payment_event for history tracking.

temperature: 0.3
output_schema: payment-analysis

tools:
  - trigger_agent
  - payment_tools.get_customer_history
  - payment_tools.record_payment_event
  - payment_tools.calculate_retry_delay

version: "1.0"

name: recovery-composer
type: llm
model: connic/gpt-5.6-terra  # or openai/gpt-5.6-sol with OpenAI BYOK configured
# Different provider so failed-payment events are not dropped during a primary outage.
fallback_model: anthropic/claude-sonnet-5
description: "Composes personalized recovery emails based on dunning stage and customer history"

system_prompt: |
  You compose recovery emails for customers with failed payments. You receive
  the dunning action details from the payment-analyzer.

  Email guidelines by stage:
  - friendly_reminder: warm, helpful tone. "We noticed an issue with your
    recent payment." Include a direct link to update payment method.
  - urgent_notice: professional but firm. Mention the specific amount and
    that service may be affected. Include payment update link.
  - final_warning: direct and clear. State the exact date service will pause.
    Offer to help if there's a billing dispute.
  - service_paused: factual, empathetic. Explain what's paused, how to
    reactivate, and that their data is safe.
  - win_back: friendly, no pressure. Mention what they're missing, offer
    to discuss pricing if cost was a factor.

  Always include:
  - Customer name (from payload)
  - Specific amount and currency
  - Clear call-to-action with payment update link placeholder: {{update_payment_url}}
  - Support contact for questions

  CRITICAL - Output format for the Email outbound connector:
  Your output schema maps directly to SMTP email fields. The "to" field
  must contain the customer's email address from the payment-analyzer's
  input. The "subject" and "body" fields become the actual email.

temperature: 0.6
output_schema: recovery-email

"""Payment and dunning tools."""

import math
from datetime import datetime, timezone
from connic.tools import db_insert, db_find


async def get_customer_history(customer_email: str) -> list[dict]:
    """Retrieve payment event history for a customer.

    Args:
        customer_email: The customer's email address.

    Returns:
        List of payment event records, most recent first.
    """
    result = await db_find(
        "payment_events",
        filter={"customer_email": customer_email},
        sort={"recorded_at": -1},
        limit=10,
    )
    return result.get("documents", [])


async def record_payment_event(
    customer_email: str,
    event_type: str,
    dunning_stage: str,
    amount_cents: int = 0,
    currency: str = "usd",
    attempt_number: int = 1,
    failure_reason: str = "",
    notes: str = "",
) -> dict:
    """Record a payment event for dunning history tracking.

    Each call inserts a new record so the full timeline is preserved.

    Args:
        customer_email: Customer's email address.
        event_type: Stripe event type (e.g. "invoice.payment_failed").
        dunning_stage: Current dunning stage (friendly_reminder, urgent_notice,
                       final_warning, service_paused, win_back).
        amount_cents: Payment amount in smallest currency unit.
        currency: ISO 4217 currency code.
        attempt_number: Which payment attempt this is.
        failure_reason: Why the payment failed (card_declined, insufficient_funds, etc.).
        notes: Additional context about the event and decision made.

    Returns:
        The inserted payment event document.
    """
    result = await db_insert("payment_events", {
        "customer_email": customer_email,
        "event_type": event_type,
        "dunning_stage": dunning_stage,
        "amount_cents": amount_cents,
        "currency": currency,
        "attempt_number": attempt_number,
        "failure_reason": failure_reason,
        "notes": notes,
        "recorded_at": datetime.now(timezone.utc).isoformat(),
    })
    return result["inserted"][0] if result.get("inserted") else result


def calculate_retry_delay(attempt_number: int, base_delay_hours: int = 24) -> dict[str, int]:
    """Calculate exponential backoff delay for payment retry.

    Uses exponential backoff capped at 7 days.

    Args:
        attempt_number: Which retry attempt this is (1-based)
        base_delay_hours: Base delay in hours before first retry

    Returns:
        Dict with delay_hours, delay_days, and attempt number
    """
    delay = min(base_delay_hours * (2 ** (attempt_number - 1)), 168)
    return {
        "delay_hours": delay,
        "delay_days": math.ceil(delay / 24),
        "attempt": attempt_number,
    }


def format_amount(amount_cents: int, currency: str = "usd") -> str:
    """Format a Stripe amount (in cents) to a human-readable string.

    Args:
        amount_cents: Amount in smallest currency unit (cents for USD/EUR)
        currency: ISO 4217 currency code

    Returns:
        Formatted amount string like "$49.99" or "49,99 EUR"
    """
    amount = amount_cents / 100
    symbols = {"usd": "$", "eur": "\u20ac", "gbp": "\u00a3"}
    symbol = symbols.get(currency.lower(), currency.upper() + " ")

    if currency.lower() in ("usd", "gbp"):
        return f"{symbol}{amount:,.2f}"
    return f"{amount:,.2f} {symbol}"

"""Middleware for payment-analyzer.

Filters irrelevant Stripe events and extracts key fields into context
so downstream agents and tools can access them.

The Stripe inbound connector delivers the full Stripe event JSON:
{id, object, type, data: {object: {...}}, created}
"""

import json
from typing import Any

from connic import StopProcessing

RELEVANT_EVENTS = {
    "invoice.payment_failed",
    "customer.subscription.updated",
    "customer.subscription.deleted",
}


async def before(content: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
    """Filter to relevant Stripe events and extract customer metadata."""
    parts = content.get("parts", [])
    for part in parts:
        if isinstance(part, dict) and "text" in part:
            try:
                payload = json.loads(part["text"])
                event_type = payload.get("type", "")
                if event_type not in RELEVANT_EVENTS:
                    raise StopProcessing(
                        f"Ignoring irrelevant Stripe event: {event_type}"
                    )
                context["stripe_event_type"] = event_type

                data_obj = payload.get("data", {}).get("object", {})
                context["stripe_customer_id"] = data_obj.get("customer", "")
                context["customer_email"] = (
                    data_obj.get("customer_email")
                    or data_obj.get("receipt_email")
                    or ""
                )
                context["customer_name"] = data_obj.get("customer_name", "")
            except (json.JSONDecodeError, AttributeError):
                pass
    return content


async def after(response: str, context: dict[str, Any]) -> str:
    return response

{
  "type": "object",
  "description": "Payment failure analysis and dunning decision",
  "properties": {
    "customer_id": {
      "type": "string",
      "description": "Stripe customer ID (cus_...)"
    },
    "customer_email": {
      "type": "string",
      "description": "Customer email address (from data.object.customer_email or receipt_email) - required for recovery email delivery"
    },
    "customer_name": {
      "type": "string",
      "description": "Customer name if available from the Stripe event"
    },
    "event_type": {
      "type": "string",
      "description": "Stripe event type that triggered this analysis"
    },
    "amount_cents": {
      "type": "integer",
      "description": "Failed payment amount in smallest currency unit (e.g. cents)"
    },
    "currency": {
      "type": "string",
      "description": "ISO 4217 currency code (e.g. usd, eur)"
    },
    "failure_reason": {
      "type": "string",
      "description": "Why the payment failed (card_declined, insufficient_funds, etc.)"
    },
    "attempt_number": {
      "type": "integer",
      "description": "Which payment attempt this is (from data.object.attempt_count)"
    },
    "dunning_action": {
      "type": "string",
      "enum": ["friendly_reminder", "urgent_notice", "final_warning", "service_paused", "win_back"],
      "description": "Decided dunning stage based on attempt_number and history"
    },
    "next_retry_hours": {
      "type": "integer",
      "description": "Hours until next automatic retry"
    }
  },
  "required": ["customer_id", "customer_email", "dunning_action", "event_type"]
}

{
  "type": "object",
  "description": "Recovery email for the Email outbound connector. Fields map directly to SMTP fields.",
  "properties": {
    "to": {
      "type": "string",
      "description": "Customer email address to send the recovery email to"
    },
    "subject": {
      "type": "string",
      "description": "Email subject line"
    },
    "body": {
      "type": "string",
      "description": "Plain text email body with {{update_payment_url}} placeholder"
    },
    "html_body": {
      "type": "string",
      "description": "Optional HTML version of the body"
    },
    "stage": {
      "type": "string",
      "enum": ["friendly_reminder", "urgent_notice", "final_warning", "service_paused", "win_back"]
    },
    "urgency": {
      "type": "string",
      "enum": ["low", "medium", "high", "critical"]
    }
  },
  "required": ["to", "subject", "body", "stage"]
}

version: "1.0"

# End-to-end through the sequential pipeline: the analyzer decides the dunning
# action (payment tools mocked, trigger_agent runs for real), then the
# recovery-composer writes the email. The final output is the recovery email.
defaults:
  runs: 1
  timeout_s: 160
  mocks: payment_mocks

tests:
  - name: failed_payment_produces_recovery_email
    payload: '{"id": "evt_3", "object": "event", "type": "invoice.payment_failed", "created": 1736500200, "data": {"object": {"customer": "cus_def456", "customer_email": "sam@example.com", "customer_name": "Sam Lee", "amount_due": 2900, "currency": "usd", "attempt_count": 2, "last_payment_error": {"code": "insufficient_funds"}}}}'
    expected_result: output.subject != "" and output.to != ""

version: "1.0"

# Custom payment tools are mocked so nothing is written to the database. The
# analyzer also calls trigger_agent (a predefined tool that runs for real), so
# strict_mocks is omitted.
defaults:
  runs: 1
  timeout_s: 120
  mocks: payment_mocks

tests:
  - name: handles_failed_payment
    payload: '{"id": "evt_1", "object": "event", "type": "invoice.payment_failed", "created": 1736500000, "data": {"object": {"customer": "cus_abc123", "customer_email": "jane@example.com", "customer_name": "Jane Doe", "amount_due": 4999, "currency": "eur", "attempt_count": 1, "last_payment_error": {"code": "card_declined"}}}}'
    expected_result: output.dunning_action != ""
    expected_tool_calls:
      - payment_tools.get_customer_history
      - payment_tools.record_payment_event
      - payment_tools.calculate_retry_delay
    expected_child_agents:
      recovery-composer:
        expected_triggered: 1

  - name: ignores_irrelevant_event
    # The before() middleware raises StopProcessing for non-dunning event types
    # before the LLM runs, so no payment tools are called.
    payload: '{"id": "evt_2", "object": "event", "type": "charge.succeeded", "created": 1736500100, "data": {"object": {"customer": "cus_xyz789"}}}'
    expected_no_tool_calls:
      - payment_tools.get_customer_history
      - payment_tools.record_payment_event
      - payment_tools.calculate_retry_delay