Webhooks

subscription = inkbox.webhooks.subscriptions.create(
    agent_identity_id=identity.id,
    url="https://example.com/events",
    event_types=[
        "message.received", "text.received", "imessage.received",
        "call.ended", "a2a.task.created",
    ],
)

const subscription = await inkbox.webhooks.subscriptions.create({
  agentIdentityId: identity.id,
  url: "https://example.com/events",
  eventTypes: [
    "message.received", "text.received", "imessage.received",
    "call.ended", "a2a.task.created",
  ],
});

inkbox.webhooks.subscriptions.create(
    agent_identity_id=identity.id,
    url="https://example.com/hook",
    event_types=["message.received", "message.bounced"],
)

await inkbox.webhooks.subscriptions.create({
      agentIdentityId: identity.id,
      url: "https://example.com/hook",
      eventTypes: ["message.received", "message.bounced"],
});

# Route incoming calls to a webhook (synchronous, response drives call)
identity.set_incoming_call_action(
    incoming_call_action="webhook",
    incoming_call_webhook_url="https://example.com/calls",
)

# Subscribe a URL to text lifecycle events (fan-out fire-and-forget)
inkbox.webhooks.subscriptions.create(
    agent_identity_id=identity.id,
    url="https://example.com/texts",
    event_types=["text.received", "text.sent", "text.delivered"],
)

import { IncomingCallAction } from "@inkbox/sdk";

// Route incoming calls to a webhook (synchronous, response drives call)
await identity.setIncomingCallAction({
      incomingCallAction: IncomingCallAction.WEBHOOK,
      incomingCallWebhookUrl: "https://example.com/calls",
});

// Subscribe a URL to text lifecycle events (fan-out fire-and-forget)
await inkbox.webhooks.subscriptions.create({
      agentIdentityId: identity.id,
      url: "https://example.com/texts",
      eventTypes: ["text.received", "text.sent", "text.delivered"],
});

inkbox.webhooks.subscriptions.create(
    agent_identity_id=identity.id,
    url="https://example.com/texts",
    event_types=[
        "text.received",
        "text.sent",
        "text.delivered",
        "text.delivery_failed",
        "text.delivery_unconfirmed",
    ],
)

await inkbox.webhooks.subscriptions.create({
      agentIdentityId: identity.id,
      url: "https://example.com/texts",
      eventTypes: [
          "text.received",
          "text.sent",
          "text.delivered",
          "text.delivery_failed",
          "text.delivery_unconfirmed",
      ],
});

# List all subscriptions on an identity
subs = inkbox.webhooks.subscriptions.list(agent_identity_id=identity.id, scope="identity")

# Update the subscription created above, not an arbitrary list result
inkbox.webhooks.subscriptions.update(
    subscription.id,
    scope="identity",
    url="https://new-host.example/hook",
    event_types=["message.received"],
)

# Delete the whole subscription, including every selected event
inkbox.webhooks.subscriptions.delete(subscription.id, scope="identity")

// List all subscriptions on an identity
const subs = await inkbox.webhooks.subscriptions.list({ agentIdentityId: identity.id, scope: "identity" });

// Update the subscription created above, not an arbitrary list result
await inkbox.webhooks.subscriptions.update(subscription.id, {
      scope: "identity",
      url: "https://new-host.example/hook",
      eventTypes: ["message.received"],
});

// Delete the whole subscription, including every selected event
await inkbox.webhooks.subscriptions.delete(subscription.id, { scope: "identity" });

# Find recent failed deliveries for a subscription
failed = inkbox.webhooks.deliveries.list(
    subscription_id=subscription.id,
    success=False,
)

# Re-send a replayable delivery to the subscription's current URL
replayable = next((delivery for delivery in failed if delivery.replayable), None)
if replayable is not None:
    inkbox.webhooks.deliveries.replay(replayable.id)

// Find recent failed deliveries for a subscription
const failed = await inkbox.webhooks.deliveries.list({
      subscriptionId: subscription.id,
      success: false,
});

// Re-send a replayable delivery to the subscription's current URL
const replayable = failed.find((delivery) => delivery.replayable === true);
if (replayable) {
      await inkbox.webhooks.deliveries.replay(replayable.id);
}

inkbox.webhooks.subscriptions.create(
    agent_identity_id=identity.id,
    url="https://example.com/imessage",
    event_types=["imessage.received", "imessage.reaction_received"],
)

await inkbox.webhooks.subscriptions.create({
      agentIdentityId: identity.id,
      url: "https://example.com/imessage",
      eventTypes: ["imessage.received", "imessage.reaction_received"],
});

inkbox.webhooks.subscriptions.create(
    agent_identity_id=identity.id,
    url="https://example.com/calls-ended",
    event_types=["call.ended"],
)

await inkbox.webhooks.subscriptions.create({
      agentIdentityId: identity.id,
      url: "https://example.com/calls-ended",
      eventTypes: ["call.ended"],
});

identity.set_incoming_call_action(
    incoming_call_action="webhook",
    incoming_call_webhook_url="https://example.com/calls",
)

import { IncomingCallAction } from "@inkbox/sdk";

await identity.setIncomingCallAction({
      incomingCallAction: IncomingCallAction.WEBHOOK,
      incomingCallWebhookUrl: "https://example.com/calls",
});

# Rotate the token
inkbox.webhooks.subscriptions.update(sub.id, auth_token="your-new-endpoint-token")

# Clear it — deliveries stop carrying the Authorization header
inkbox.webhooks.subscriptions.update(sub.id, auth_token=None)

// Rotate the token
await inkbox.webhooks.subscriptions.update(sub.id, {
  authToken: "your-new-endpoint-token",
});

// Clear it — deliveries stop carrying the Authorization header
await inkbox.webhooks.subscriptions.update(sub.id, { authToken: null });

# Rotate the token, read from stdin so it stays out of shell history
printf '%s' "$NEW_ENDPOINT_TOKEN" | inkbox webhook subscription update SUB_ID --auth-token-stdin

# Clear it
inkbox webhook subscription update SUB_ID --clear-auth-token

from inkbox import verify_webhook

# FastAPI
@app.post("/hooks/mail")
async def mail_hook(request: Request):
    raw_body = await request.body()
    if not verify_webhook(
        payload=raw_body,
        headers=request.headers,
        secret="whsec_...",
    ):
        raise HTTPException(status_code=403)
    ...

# Flask
@app.post("/hooks/mail")
def mail_hook():
    raw_body = request.get_data()
    if not verify_webhook(
        payload=raw_body,
        headers=request.headers,
        secret="whsec_...",
    ):
        abort(403)
    ...

import { verifyWebhook } from "@inkbox/sdk";

// Express — use express.raw() to get the raw body Buffer
app.post("/hooks/mail", express.raw({ type: "*/*" }), (req, res) => {
      const valid = verifyWebhook({
          payload: req.body,
          headers: req.headers,
          secret: "whsec_...",
      });
      if (!valid) return res.status(403).end();
      // handle event ...
});

import json
from typing import cast

from inkbox import MailWebhookPayload, verify_webhook

@app.post("/hooks/mail")
async def mail_hook(request: Request):
    raw_body = await request.body()
    if not verify_webhook(
        payload=raw_body,
        headers=request.headers,
        secret="whsec_...",
    ):
        raise HTTPException(status_code=403)

    payload = cast(MailWebhookPayload, json.loads(raw_body))
    msg = payload["data"]["message"]
    contacts = payload["data"]["contacts"]              # always present
    agent_identities = payload["data"]["agent_identities"]  # always present

    # Inbound: look up the sender (and any internal-agent CC peers)
    if payload["event_type"] == "message.received":
        sender = next(
            (c for c in contacts if c["bucket"] == "from"),
            None,
        )
        if sender is not None:
            logger.info("inbound from known contact %s", sender["id"])
        for ai in agent_identities:
            logger.info(
                "internal-agent peer %s @%s (bucket=%s)",
                ai["id"], ai["agent_handle"], ai["bucket"],
            )

    # Outbound: bcc_addresses is populated; surface every matched recipient
    if msg["direction"] == "outbound":
        for c in contacts:
            logger.info(
                "matched %s recipient %s -> contact %s",
                c["bucket"], c["address"], c["id"],
            )

import { MailWebhookPayload, verifyWebhook } from "@inkbox/sdk";

app.post("/hooks/mail", express.raw({ type: "*/*" }), (req, res) => {
      if (
          !verifyWebhook({
              payload: req.body,
              headers: req.headers,
              secret: "whsec_...",
          })
      ) {
          return res.status(403).end();
      }
      const payload = JSON.parse(req.body.toString()) as MailWebhookPayload;
      const msg = payload.data.message;
      const contacts = payload.data.contacts;                  // always present
      const agentIdentities = payload.data.agent_identities;   // always present

      if (payload.event_type === "message.received") {
          const sender = contacts.find((c) => c.bucket === "from");
          if (sender) {
              console.log("inbound from known contact", sender.id);
          }
          for (const ai of agentIdentities) {
              console.log(
                  `internal-agent peer ${ai.id} @${ai.agent_handle} (bucket=${ai.bucket})`,
              );
          }
      }

      if (msg.direction === "outbound") {
          for (const c of contacts) {
              console.log(
                  `matched ${c.bucket} recipient ${c.address} -> contact ${c.id}`,
              );
          }
      }

      res.status(204).end();
});

import json
from typing import cast

from inkbox import TextWebhookPayload, verify_webhook

@app.post("/hooks/text")
async def text_hook(request: Request):
    raw_body = await request.body()
    if not verify_webhook(
        payload=raw_body,
        headers=request.headers,
        secret="whsec_...",
    ):
        raise HTTPException(status_code=403)

    payload = cast(TextWebhookPayload, json.loads(raw_body))
    data = payload["data"]
    msg = data["text_message"]

    # Group outbound: per-recipient lifecycle. data.recipient_phone_number
    # identifies which leg this event is for; per-leg state lives in
    # text_message.recipients[].
    is_group_outbound = (
        msg["direction"] == "outbound" and msg["remote_phone_number"] is None
    )
    if is_group_outbound:
        leg = data["recipient_phone_number"]
        for r in msg["recipients"] or []:
            if r["recipient_phone_number"] == leg:
                logger.info(
                    "leg %s status=%s error=%s",
                    leg, r["delivery_status"], r["error_code"],
                )

    match payload["event_type"]:
        case "text.delivery_failed":
            msg = payload["data"]["text_message"]
            recipient = (
                payload["data"]["recipient_phone_number"]
                or msg["remote_phone_number"]
            )
            receipt = next(
                (
                    r for r in (msg["recipients"] or [])
                    if r["recipient_phone_number"] == recipient
                ),
                None,
            )
            logger.error(
                "text to %s failed: %s (%s)",
                recipient,
                (receipt or msg)["error_code"],
                (receipt or msg)["error_detail"],
            )
        case "text.delivered":
            # For group events, check payload["data"]["recipient_phone_number"].
            ...
        case "text.received":
            for c in data["contacts"]:
                logger.info("inbound from known contact %s", c["id"])
            for ai in data["agent_identities"]:
                logger.info("inbound from internal agent @%s", ai["agent_handle"])

import { TextWebhookPayload, verifyWebhook } from "@inkbox/sdk";

app.post("/hooks/text", express.raw({ type: "*/*" }), (req, res) => {
  if (
      !verifyWebhook({
          payload: req.body,
          headers: req.headers,
          secret: "whsec_...",
      })
  ) {
      return res.status(403).end();
  }
  const payload = JSON.parse(req.body.toString()) as TextWebhookPayload;
  switch (payload.event_type) {
      case "text.delivery_failed": {
          const m = payload.data.text_message;
          const recipient =
              payload.data.recipient_phone_number ?? m.remote_phone_number;
          const receipt = m.recipients?.find(
              (r) => r.recipient_phone_number === recipient,
          );
          console.error(
              `text to ${recipient} failed`,
              receipt?.error_code ?? m.error_code,
              receipt?.error_detail ?? m.error_detail,
          );
          break;
      }
      case "text.delivered":
          // For group events, check payload.data.recipient_phone_number.
          break;
      case "text.received":
          for (const c of payload.data.contacts) {
              console.log("inbound from known contact", c.id);
          }
          for (const ai of payload.data.agent_identities) {
              console.log("inbound from internal agent @" + ai.agent_handle);
          }
      }

      switch (payload.event_type) {
          case "text.delivery_failed":
              console.error(
                  `SMS to ${msg.remote_phone_number} failed`,
                  msg.error_code,
                  msg.error_detail,
              );
              break;
          case "text.received":
              for (const c of data.contacts) {
                  console.log("inbound from known contact", c.id);
              }
              for (const ai of data.agent_identities) {
                  console.log("inbound from internal agent @" + ai.agent_handle);
              }
              break;
      }
      res.status(204).end();
});

import json
from typing import cast

from inkbox import PhoneIncomingCallWebhookPayload, verify_webhook

@app.post("/hooks/calls")
async def call_hook(request: Request):
    raw_body = await request.body()
    if not verify_webhook(
        payload=raw_body,
        headers=request.headers,
        secret="whsec_...",
    ):
        raise HTTPException(status_code=403)

    payload = cast(PhoneIncomingCallWebhookPayload, json.loads(raw_body))
    for c in payload["contacts"]:
        logger.info("inbound call from known contact %s", c["id"])
    for ai in payload["agent_identities"]:
        logger.info("inbound call from internal agent @%s", ai["agent_handle"])

    # Respond with how Inkbox should handle the call
    return {"action": "answer", "client_websocket_url": "wss://..."}

import { PhoneIncomingCallWebhookPayload, verifyWebhook } from "@inkbox/sdk";

app.post("/hooks/calls", express.raw({ type: "*/*" }), (req, res) => {
      if (
          !verifyWebhook({
              payload: req.body,
              headers: req.headers,
              secret: "whsec_...",
          })
      ) {
          return res.status(403).end();
      }
      const payload = JSON.parse(req.body.toString()) as PhoneIncomingCallWebhookPayload;
      for (const c of payload.contacts) {
          console.log("inbound call from known contact", c.id);
      }
      for (const ai of payload.agent_identities) {
          console.log("inbound call from internal agent @" + ai.agent_handle);
      }

      // Respond with how Inkbox should handle the call
      res.json({ action: "answer", client_websocket_url: "wss://..." });
});

import json
from typing import cast

from inkbox import CallEndedWebhookPayload, verify_webhook

@app.post("/hooks/calls-ended")
async def call_ended_hook(request: Request):
    raw_body = await request.body()
    if not verify_webhook(
        payload=raw_body,
        headers=request.headers,
        secret="whsec_...",
    ):
        raise HTTPException(status_code=403)

    payload = cast(CallEndedWebhookPayload, json.loads(raw_body))
    if payload["event_type"] != "call.ended":
        return

    data = payload["data"]
    call = data["call"]
    logger.info(
        "call %s ended: %s, %ss",
        call["id"], call["status"], call["duration_seconds"],
    )

    # Inline transcript is present when the call has transcribed turns.
    transcript = data["transcript"]
    if transcript is not None:
        for entry in transcript["entries"]:
            if entry.get("marker") == "abridged":
                continue  # skip the middle-cut marker
            logger.info("[%s] %s", entry["party"], entry["text"])

    # transcript_url is always present and authoritative (needs an API key).
    logger.info("full transcript at %s", data["transcript_url"])

import { CallEndedWebhookPayload, verifyWebhook } from "@inkbox/sdk";

app.post("/hooks/calls-ended", express.raw({ type: "*/*" }), (req, res) => {
      if (
          !verifyWebhook({
              payload: req.body,
              headers: req.headers,
              secret: "whsec_...",
          })
      ) {
          return res.status(403).end();
      }
      const payload = JSON.parse(req.body.toString()) as CallEndedWebhookPayload;
      if (payload.event_type !== "call.ended") return res.status(204).end();

      const { call, transcript, transcript_url } = payload.data;
      console.log(`call ${call.id} ended: ${call.status}, ${call.duration_seconds}s`);

      // Inline transcript is present when the call has transcribed turns.
      if (transcript) {
          for (const entry of transcript.entries) {
              if (entry.marker === "abridged") continue; // skip the middle-cut marker
              console.log(`[${entry.party}] ${entry.text}`);
          }
      }

      // transcript_url is always present and authoritative (needs an API key).
      console.log("full transcript at", transcript_url);
      res.status(204).end();
});