WebhookReceiver - Autoplay SDK

WebhookReceiver is for customers using push mode: when your product is configured with a forward_url, the connector POSTs events to your endpoint as JSON. WebhookReceiver handles signature verification, JSON parsing, and dispatching to typed callbacks — giving you the same ActionsPayload / SummaryPayload interface that SSE users get from AsyncConnectorClient. Use it when you want the connector to push events to your server rather than your server pulling from the SSE stream.

How push mode works

Autoplay connector  ──HTTP POST──►  your endpoint
                                        ↓
                               WebhookReceiver.handle()
                                        ↓
                              ActionsPayload / SummaryPayload
                                        ↓
                               your callback

Every POST includes an X-Connector-Signature header for verification:

X-Connector-Signature: sha256=<HMAC-SHA256(key=secret, message=raw_body)>

Setup (FastAPI)

import os
from fastapi import FastAPI, Header, HTTPException, Request
from autoplay_sdk import WebhookReceiver, ActionsPayload, SummaryPayload

app = FastAPI()

async def handle_actions(payload: ActionsPayload) -> None:
    for action in payload.actions:
        embedding = await embed(action.to_text())
        await vector_store.upsert(id=payload.session_id, vector=embedding)

async def handle_summary(payload: SummaryPayload) -> None:
    await vector_store.upsert(id=payload.session_id, vector=await embed(payload.summary))

receiver = WebhookReceiver(
    secret=os.getenv("CONNECTOR_SECRET", ""),
    on_actions=handle_actions,
    on_summary=handle_summary,
)

@app.post("/events")
async def events(
    request: Request,
    x_connector_signature: str | None = Header(default=None),
):
    body = await request.body()
    try:
        await receiver.handle(body, x_connector_signature)
    except ValueError:
        raise HTTPException(status_code=401, detail="Invalid signature")
    return {"status": "ok"}

Setup (Flask / sync)

import os
from flask import Flask, request, abort
from autoplay_sdk import WebhookReceiver, ActionsPayload

app = Flask(__name__)

def handle_actions(payload: ActionsPayload) -> None:
    for action in payload.actions:
        print(action.title, action.description)

receiver = WebhookReceiver(
    secret=os.getenv("CONNECTOR_SECRET", ""),
    on_actions=handle_actions,
)

@app.post("/events")
def events():
    body = request.get_data()
    sig  = request.headers.get("X-Connector-Signature")
    try:
        receiver.handle_sync(body, sig)
    except ValueError:
        abort(401)
    return {"status": "ok"}

Constructor

WebhookReceiver(
    secret="",
    on_actions=None,
    on_summary=None,
)

secret

string

default:"\"\""

HMAC-SHA256 signing secret configured on the connector (integration_config.secret). Pass an empty string to skip verification — useful in development but not recommended in production.

on_actions

Callable[[ActionsPayload], None]

default:"None"

Sync or async callback called for every actions event. Can also be registered via .on_actions(fn) after construction.

on_summary

Callable[[SummaryPayload], None]

default:"None"

Sync or async callback called for every summary event. Can also be registered via .on_summary(fn) after construction.

Builder interface

Both callback registration methods return self for chaining:

receiver = (
    WebhookReceiver(secret=SECRET)
    .on_actions(handle_actions)
    .on_summary(handle_summary)
)

Core methods

`await .handle(body, signature_header)` — async

Verify, parse, and dispatch. Use with FastAPI, Starlette, aiohttp, or any async framework.

payload = await receiver.handle(body, x_connector_signature)

Raises ValueError if the signature is invalid. Returns the parsed ActionsPayload or SummaryPayload, or None if the event type is unrecognised.

`.handle_sync(body, signature_header)` — sync

Same behaviour but synchronous. Use with Flask, Django, or any sync framework. Callbacks must be synchronous.

payload = receiver.handle_sync(body, sig)

`.verify(body, signature_header)` → `bool`

Check the HMAC-SHA256 signature without parsing or dispatching.

if not receiver.verify(body, sig):
    return 401

`.parse(body)` → `ActionsPayload | SummaryPayload | None`

Parse raw bytes into a typed payload without verifying the signature or dispatching callbacks.

payload = receiver.parse(body)
if isinstance(payload, ActionsPayload):
    print(payload.session_id, payload.count)

API reference

Method	Returns	Description
`.on_actions(fn)`	`self`	Register actions callback (sync or async)
`.on_summary(fn)`	`self`	Register summary callback (sync or async)
`await .handle(body, sig)`	`AnyPayload \| None`	Verify + parse + dispatch (async)
`.handle_sync(body, sig)`	`AnyPayload \| None`	Verify + parse + dispatch (sync)
`.verify(body, sig)`	`bool`	HMAC signature check only
`.parse(body)`	`AnyPayload \| None`	Parse bytes to typed payload, no verification

​How push mode works

​Setup (FastAPI)

​Setup (Flask / sync)

​Constructor

​Builder interface

​Core methods

​await .handle(body, signature_header) — async

​.handle_sync(body, signature_header) — sync

​.verify(body, signature_header) → bool

​.parse(body) → ActionsPayload | SummaryPayload | None

​API reference