# Stripe Webhook → Anthropic Tool Pattern for Async Agents

> Convert Stripe webhook events (payment, dispute, refund) into Claude tool calls so agents react to real-time payment lifecycle events.

## Install

Save the content below to `.claude/skills/` or append to your `CLAUDE.md`:

## Quick Use

1. Add a Stripe webhook endpoint at dashboard.stripe.com/webhooks pointing at /stripe/webhook
2. Set STRIPE_WEBHOOK_SECRET, STRIPE_SECRET_KEY, ANTHROPIC_API_KEY
3. Map events to tool names in EVENT_TO_TOOL

---

## Intro

This skill turns Stripe webhook events into Anthropic tool-call inputs so a Claude agent reacts to real payment lifecycle events: payment succeeded, dispute opened, subscription canceled, refund issued. The pattern is a tiny FastAPI/Express endpoint that verifies the Stripe signature, transforms the event into a structured tool call, and dispatches the call against your existing agent. Best for: support agents that triage disputes, billing chatbots that confirm payments, churn-intervention agents reacting to subscription events. Works with: Anthropic API, OpenAI Tool Use, any agent that takes tool inputs. Setup time: 15 minutes.

---

### FastAPI endpoint

```python
import os, stripe
from anthropic import Anthropic
from fastapi import FastAPI, Request, HTTPException

app = FastAPI()
stripe.api_key = os.environ["STRIPE_SECRET_KEY"]
WEBHOOK_SECRET = os.environ["STRIPE_WEBHOOK_SECRET"]
claude = Anthropic()

EVENT_TO_TOOL = {
    "charge.dispute.created":         "handle_dispute",
    "invoice.payment_failed":         "retry_failed_payment",
    "customer.subscription.deleted":  "churn_outreach",
    "charge.refunded":                "log_refund",
}

@app.post("/stripe/webhook")
async def webhook(request: Request):
    payload = await request.body()
    sig = request.headers.get("stripe-signature")
    try:
        event = stripe.Webhook.construct_event(payload, sig, WEBHOOK_SECRET)
    except (ValueError, stripe.error.SignatureVerificationError):
        raise HTTPException(400, "bad signature")

    tool_name = EVENT_TO_TOOL.get(event["type"])
    if not tool_name:
        return {"status": "ignored"}

    obj = event["data"]["object"]
    msg = claude.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=1024,
        tools=[{
            "name": tool_name,
            "description": f"Triggered by Stripe event {event['type']}",
            "input_schema": {"type": "object", "properties": {
                "stripe_id": {"type": "string"},
                "amount":    {"type": "integer"},
                "customer":  {"type": "string"},
                "reason":    {"type": "string"},
            }},
        }],
        messages=[{
            "role": "user",
            "content": f"A Stripe {event['type']} event occurred. Decide on next action.\n\n{obj}",
        }],
    )
    return {"status": "dispatched", "claude_response": msg.content}
```

### Production hardening

- **Signature verification is mandatory** — never skip `construct_event`. Without it, anyone can POST fake events.
- **Idempotency** — Stripe retries webhooks for 3 days on non-2xx. Store `event["id"]` in Redis with 7-day TTL; skip if seen.
- **Async dispatch** — return 200 within 10s. Push the Claude call to a queue (Celery / Inngest / SQS) if it takes longer.
- **Replay attacks** — `construct_event` enforces 5-minute timestamp tolerance. Don't widen it.

---

### FAQ

**Q: Why not have Claude call Stripe directly with the Agent Toolkit?**
A: For synchronous user-driven flows, do that. This pattern is for async events that happen without a user prompt — disputes opened by the cardholder, subscriptions canceled by Stripe Smart Retries, payments failing on retry. The webhook is the trigger; Claude is the policy engine.

**Q: How do I scope which events fire Claude?**
A: Whitelist in the EVENT_TO_TOOL map. Stripe will keep delivering all subscribed events but you only spend Claude tokens on the ones you map. Subscribe selectively at dashboard.stripe.com/webhooks too.

**Q: What about test mode?**
A: Use Stripe CLI: `stripe trigger charge.dispute.created --forward-to localhost:8000/stripe/webhook`. The CLI signs events with your test webhook secret and replays them locally.

---

## Source & Thanks

> Pattern compiled from [Stripe Webhooks](https://docs.stripe.com/webhooks) + [Anthropic Tool Use](https://docs.anthropic.com/en/docs/tool-use).
>
> Stripe MIT-licensed, Anthropic SDK MIT-licensed.

---

<!-- ZH -->

## 快速使用

1. 在 dashboard.stripe.com/webhooks 加 webhook endpoint 指向 /stripe/webhook
2. 设 STRIPE_WEBHOOK_SECRET、STRIPE_SECRET_KEY、ANTHROPIC_API_KEY
3. 在 EVENT_TO_TOOL 里把事件映射到 tool 名

---

## 简介

这个 skill 把 Stripe webhook 事件转成 Anthropic tool-call 输入，让 Claude agent 对真实支付生命周期事件做反应：支付成功、争议打开、订阅取消、退款发起。模式是一个小 FastAPI/Express endpoint —— 验证 Stripe 签名、把事件转成结构化 tool call、分派给现有 agent。适合处理争议的客服 agent、确认支付的计费 chatbot、对订阅事件做反应的流失干预 agent。兼容 Anthropic API、OpenAI Tool Use、任何接 tool 输入的 agent。装机时间 15 分钟。

---

### FastAPI endpoint

```python
import os, stripe
from anthropic import Anthropic
from fastapi import FastAPI, Request, HTTPException

app = FastAPI()
stripe.api_key = os.environ["STRIPE_SECRET_KEY"]
WEBHOOK_SECRET = os.environ["STRIPE_WEBHOOK_SECRET"]
claude = Anthropic()

EVENT_TO_TOOL = {
    "charge.dispute.created":         "handle_dispute",
    "invoice.payment_failed":         "retry_failed_payment",
    "customer.subscription.deleted":  "churn_outreach",
    "charge.refunded":                "log_refund",
}

@app.post("/stripe/webhook")
async def webhook(request: Request):
    payload = await request.body()
    sig = request.headers.get("stripe-signature")
    try:
        event = stripe.Webhook.construct_event(payload, sig, WEBHOOK_SECRET)
    except (ValueError, stripe.error.SignatureVerificationError):
        raise HTTPException(400, "bad signature")

    tool_name = EVENT_TO_TOOL.get(event["type"])
    if not tool_name:
        return {"status": "ignored"}

    obj = event["data"]["object"]
    msg = claude.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=1024,
        tools=[{
            "name": tool_name,
            "description": f"Triggered by Stripe event {event['type']}",
            "input_schema": {"type": "object", "properties": {
                "stripe_id": {"type": "string"},
                "amount":    {"type": "integer"},
                "customer":  {"type": "string"},
                "reason":    {"type": "string"},
            }},
        }],
        messages=[{
            "role": "user",
            "content": f"A Stripe {event['type']} event occurred. Decide on next action.\n\n{obj}",
        }],
    )
    return {"status": "dispatched", "claude_response": msg.content}
```

### 生产加固

- **签名验证必须做** —— 别跳过 `construct_event`。不验证的话任何人都能 POST 假事件。
- **幂等** —— Stripe 对非 2xx 响应重试 3 天。把 `event["id"]` 存 Redis 7 天 TTL，见过就跳。
- **异步分派** —— 10 秒内必须返回 200。Claude 调用慢的话推到队列（Celery / Inngest / SQS）。
- **重放攻击** —— `construct_event` 强制 5 分钟时间戳容差。别放宽。

---

### FAQ

**Q: 为啥不让 Claude 直接用 Agent Toolkit 调 Stripe？**
A: 同步、用户驱动的流程就那么干。这个模式给异步事件用 —— 持卡人发起的争议、Stripe Smart Retry 取消的订阅、重试失败的支付，都没用户 prompt。Webhook 是触发器，Claude 是策略引擎。

**Q: 怎么控制哪些事件触发 Claude？**
A: 在 EVENT_TO_TOOL 映射里白名单。Stripe 会继续投递所有订阅事件，但只对映射的事件花 Claude token。也可以在 dashboard.stripe.com/webhooks 选择性订阅。

**Q: 测试模式怎么搞？**
A: 用 Stripe CLI：`stripe trigger charge.dispute.created --forward-to localhost:8000/stripe/webhook`。CLI 用测试 webhook secret 签事件，本地回放。

---

## 来源与感谢

> Pattern compiled from [Stripe Webhooks](https://docs.stripe.com/webhooks) + [Anthropic Tool Use](https://docs.anthropic.com/en/docs/tool-use).
>
> Stripe MIT-licensed, Anthropic SDK MIT-licensed.


---
Source: https://tokrepo.com/en/workflows/stripe-webhook-anthropic-tool-pattern-for-async-agents
Author: Stripe