Set up transaction alerts with webhooks

Build a service that notifies you when a transaction exceeds a threshold, matches a specific category, or hits a particular account — all powered by Yoshi webhooks.

Register a webhook endpoint

Create an endpoint to receive transaction.created events:

curl -X POST https://api.yoshi.ai/v1/webhooks/endpoints \
  -H "Authorization: Bearer yoshi_YOUR_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "url": "https://your-server.com/webhooks/yoshi",
    "filter_types": ["transaction.created"]
  }'

You can also manage endpoints through the self-service portal.

Build the webhook receiver

Set up an Express server that verifies the webhook signature and filters transactions:

import express from "express";
import { Webhook } from "svix";

const app = express();
app.use(express.raw({ type: "application/json" }));

const ALERT_THRESHOLD_CENTS = 10000; // $100.00

app.post("/webhooks/yoshi", async (req, res) => {
  // Verify the signature first
  const wh = new Webhook(process.env.WEBHOOK_SECRET);
  let event;
  try {
    event = wh.verify(req.body, {
      "webhook-id": req.headers["webhook-id"],
      "webhook-timestamp": req.headers["webhook-timestamp"],
      "webhook-signature": req.headers["webhook-signature"],
    });
  } catch (err) {
    return res.status(400).send("Invalid signature");
  }

  // Acknowledge immediately, process async
  res.sendStatus(200);

  if (event.type === "transaction.created") {
    await processTransactionEvent(event.data);
  }
});

app.listen(3000);

Filter and alert

Fetch the full transaction details using the IDs from the webhook payload, then apply your filtering logic:

import Yoshi from "@yoshi-ai/sdk";

const yoshi = new Yoshi();

async function processTransactionEvent(data) {
  const transactions = await yoshi.transactions.list({
    account_id: data.account_id,
    limit: data.count,
  });

  for (const tx of transactions.data) {
    // Filter by amount
    if (tx.amount_absolute >= ALERT_THRESHOLD_CENTS / 100) {
      await sendAlert({
        message: `Large transaction: $${tx.amount_absolute} at ${tx.counterparty_name}`,
        account: tx.account_name,
        category: tx.category_label,
      });
    }

    // Filter by category
    if (tx.category_tier1 === "Travel") {
      await sendAlert({
        message: `Travel expense: $${tx.amount_absolute} at ${tx.counterparty_name}`,
      });
    }
  }
}

Send to Slack

Post alerts to a Slack channel using an incoming webhook:

async function sendAlert({ message, account, category }) {
  await fetch(process.env.SLACK_WEBHOOK_URL, {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({
      text: `🔔 *Transaction Alert*\n${message}`,
      blocks: [
        {
          type: "section",
          text: {
            type: "mrkdwn",
            text: [
              `*${message}*`,
              account ? `Account: ${account}` : null,
              category ? `Category: ${category}` : null,
            ]
              .filter(Boolean)
              .join("\n"),
          },
        },
      ],
    }),
  });
}

Handle duplicates

Webhook events can be delivered more than once. Use the event id for deduplication:

const processedEvents = new Set();

app.post("/webhooks/yoshi", async (req, res) => {
  const event = verifyWebhook(req);
  res.sendStatus(200);

  if (processedEvents.has(event.id)) {
    return; // Already processed
  }
  processedEvents.add(event.id);

  await processTransactionEvent(event.data);
});

For production use, store processed event IDs in a database or Redis instead of an in-memory Set.

Python example

from fastapi import FastAPI, Request, HTTPException
from svix.webhooks import Webhook
import os, httpx

app = FastAPI()
THRESHOLD = 100.00  # dollars

@app.post("/webhooks/yoshi")
async def handle_webhook(request: Request):
    body = await request.body()
    headers = dict(request.headers)

    wh = Webhook(os.environ["WEBHOOK_SECRET"])
    try:
        event = wh.verify(body, headers)
    except Exception:
        raise HTTPException(status_code=400, detail="Invalid signature")

    if event["type"] == "transaction.created":
        await process_transactions(event["data"])

    return {"status": "ok"}

async def process_transactions(data):
    async with httpx.AsyncClient() as client:
        response = await client.get(
            "https://api.yoshi.ai/transactions",
            params={"account_id": data["account_id"], "limit": data["count"]},
            headers={"Authorization": f"Bearer {os.environ['YOSHI_API_KEY']}"},
        )
        for tx in response.json()["data"]:
            if tx["amount_absolute"] >= THRESHOLD:
                print(f"ALERT: ${tx['amount_absolute']} at {tx['counterparty_name']}")

Set up transaction alerts with webhooks

Register a webhook endpoint

Build the webhook receiver

Filter and alert

Send to Slack

Handle duplicates

Python example

What’s next

Event catalog

Signature verification

​Register a webhook endpoint

​Build the webhook receiver

​Filter and alert

​Send to Slack

​Handle duplicates

​Python example

​What’s next

Event catalog

Signature verification

Register a webhook endpoint

Build the webhook receiver

Filter and alert

Send to Slack

Handle duplicates

Python example

What’s next