Webhooks notify your app when something happens in Cyberdesk. Instead of polling, subscribe to the event types you care about (see the Event Catalog in the Webhooks portal). The most common one is run_complete, which we’ll use below as an example.
Typical flow: create a run with the SDK, return immediately to your user, and update your system when you receive the webhook.
import express from "express";import { Webhook, WebhookVerificationError } from "svix";import type { RunCompletedEvent, RunResponse } from "cyberdesk";const app = express();app.use(express.raw({ type: "application/json" })); // verify requires raw bodyfunction assertRunCompletedEvent(x: unknown): asserts x is RunCompletedEvent { if (!x || (x as any).event_type !== "run_complete" || !(x as any).run) { throw new Error("Invalid run_complete payload"); }}app.post("/webhooks/cyberdesk", (req, res) => { const secret = process.env.SVIX_WEBHOOK_SECRET!; const wh = new Webhook(secret); const headers = { "svix-id": req.header("svix-id")!, "svix-timestamp": req.header("svix-timestamp")!, "svix-signature": req.header("svix-signature")!, }; try { const payload = wh.verify(req.body, headers); assertRunCompletedEvent(payload); // Narrowed type: payload is RunCompletedEvent; run is fully typed const run: RunResponse = payload.run; // idempotency: upsert on headers["svix-id"] or payload.event_id // handle success/error/cancelled res.status(200).end(); } catch (err) { if (err instanceof WebhookVerificationError) return res.status(400).end(); res.status(500).end(); }});
import { createCyberdeskClient } from "cyberdesk";import type { RunResponse } from "cyberdesk";async function onRunComplete(run: RunResponse) { const client = createCyberdeskClient(process.env.CYBERDESK_API_KEY!); // Example A: Start a follow-up workflow await client.runs.create({ workflow_id: process.env.NEXT_WORKFLOW_ID!, input_values: { summary: run.output_data?.summary } }); // Example B: Persist and enqueue for async processing await db.runs.upsert({ id: run.id, status: run.status, output: run.output_data }); await queue.enqueue("postprocess-run", { runId: run.id });}
from fastapi import FastAPI, Request, HTTPExceptionfrom svix.webhooks import Webhook, WebhookVerificationErrorfrom openapi_client.cyberdesk_cloud_client.models.run_completed_event import RunCompletedEventfrom openapi_client.cyberdesk_cloud_client.models.run_response import RunResponsefrom typing import castimport osapp = FastAPI()@app.post("/webhooks/cyberdesk")async def cyberdesk_webhook(request: Request): secret = os.environ["SVIX_WEBHOOK_SECRET"] wh = Webhook(secret) payload = await request.body() headers = { "svix-id": request.headers.get("svix-id"), "svix-timestamp": request.headers.get("svix-timestamp"), "svix-signature": request.headers.get("svix-signature"), } try: data = wh.verify(payload, headers) evt = RunCompletedEvent.from_dict(data) # Help some IDEs narrow the type for evt.run run: RunResponse = cast(RunResponse, evt.run) # idempotency: upsert on headers["svix-id"] or evt.event_id return {"ok": True} except WebhookVerificationError: raise HTTPException(status_code=400, detail="Invalid signature")
from cyberdesk import CyberdeskClient, RunCreatefrom openapi_client.cyberdesk_cloud_client.models.run_response import RunResponseasync def on_run_complete(run: RunResponse): client = CyberdeskClient(os.environ["CYBERDESK_API_KEY"]) # Handle different statuses if run.status == "success": # Example A: Start a follow-up workflow follow_up = RunCreate( workflow_id=os.environ["NEXT_WORKFLOW_ID"], input_values={"summary": (run.output_data or {}).get("summary")} ) await client.runs.create(follow_up) # Example B: Persist and enqueue await db.upsert_run(run.id, run.status, run.output_data) await queue.enqueue("postprocess-run", {"run_id": run.id}) elif run.status == "error": # Handle errors await notify_error(run.id, run.error)
Type safety:
TypeScript: use a type guard (like assertRunCompletedEvent) to narrow the Svix‑verified payload to RunCompletedEvent, then type run as RunResponse.
Python: RunCompletedEvent.from_dict(data) returns a typed attrs object; some IDEs may still show Any | RunResponse for evt.run, so cast(RunResponse, evt.run) helps IDEs while remaining safe after validation. Note: The SDK uses attrs classes, not Pydantic, so use from_dict() not model_validate().
Use run.status to branch your logic and output_data or attachments to continue your process.
Sensitive inputs: plaintext secrets are never included in webhook payloads. If a run used sensitive variables, you’ll see run.sensitive_input_aliases, which maps each sensitive input key (for example, password) to the secure secret identifier used during execution. Actual values stay in the secure vault and are deleted after completion.
