Python

When an AI agent reaches an action that needs sign-off, it opens an interrupt - a request for a human decision. A teammate handles it in the inbox, and your code continues with the outcome. In between, a session streams the agent's conversation to Live View so operators can watch it as it happens. vigilator-py-sdk is the official Python SDK for both:

Installation

uv add vigilator-py-sdk

Quick Start

from vigilator_py_sdk import Client

client = Client(key="your-api-key")

from vigilator_py_sdk import ActionRequest, AllowedDecision, InterruptsPostRequest

interrupt = client.create_interrupt(
    InterruptsPostRequest(
        title="Send onboarding email",
        description="The agent wants to email a new customer.",
        externalId="run_42",  # your own correlation id, e.g. the agent run
        actionRequests=[
            ActionRequest(
                name="send_email",
                args={"to": "customer@example.com"},
                allowedDecisions=[AllowedDecision.approve, AllowedDecision.edit, AllowedDecision.reject],
            )
        ],
    )
)

result = client.get_interrupt(interrupt.id)
if result.answered:
    for request in result.actionRequests:
        print(request.name, request.decision, request.decidedByName)

Acting on decisions

Each decided action request tells you what the reviewer chose and everything you need to act on it:

Every decision also carries decidedByName and decidedAt, so your agent can log who made the call and when.

Client behaviour

The client retries requests that fail with 429/502/503/504 or a connection error - 5 attempts with exponential backoff by default, tunable via the constructor. Use it as a context manager to release the connection pool when you are done:

with Client(key="your-api-key") as client:
    ...

Live View sessions

Interrupts capture the moments an agent stops to ask; a session covers everything in between. Register one when a run starts, stream the conversation as it progresses, and end it when the run completes - operators watch the transcript grow in Live View and can step in when something looks off. Sessions are standalone: they are not linked to interrupts, so you can monitor agents that never raise one.

from vigilator_py_sdk import Message, MessageType

session = client.start_session(
    "billing-agent",
    external_id="thread_42",
    messages=[Message(type=MessageType.human, content="Refund order #42")],
)

client.append_session_messages(
    session.id,
    [
        Message(type=MessageType.ai, content="Looking up order #42.", name="billing-agent"),
        Message(type=MessageType.ai, content="Refund issued.", name="billing-agent"),
    ],
)

client.end_session(session.id)

A session that goes quiet for longer than your organisation's session timeout is ended automatically, so a crashed agent never lingers in the queue. If your run may pause for long stretches - waiting on an interrupt, say - end the session before the pause and start a new one afterwards, or make sure the timeout in Integrations → Live View covers it.

Session state

Each call returns the session as the API sees it:

Monitored time counts towards your organisation's agent hours, recorded when each session ends. Once a free organisation's monthly allowance is spent, start_session raises UsageLimitError until it resets.

Handling webhooks

Vigilator signs webhook deliveries (Svix / Standard Webhooks). WebhookHandler verifies the signature, parses the payload into a typed event, and dispatches it to registered callbacks. Create an endpoint in the dashboard under your organisation's integrations, copy its signing secret, and register callbacks for the events you care about:

from vigilator_py_sdk import InterruptAnsweredEvent, WebhookHandler

webhooks = WebhookHandler(secret="whsec_...")  # the endpoint's signing secret

@webhooks.on("interrupt.answered")
def resume_agent(event: InterruptAnsweredEvent) -> None:
    for request in event.data.action_requests:
        ...  # act on approve / edit / reject / respond

The handler is framework-agnostic - pass the raw request body and the request headers from any web framework:

from fastapi import FastAPI, HTTPException, Request, Response
from vigilator_py_sdk import WebhookVerificationError

app = FastAPI()

@app.post("/webhooks/vigilator")
async def vigilator_webhook(request: Request) -> Response:
    try:
        webhooks.handle(await request.body(), request.headers)
    except WebhookVerificationError:
        raise HTTPException(status_code=401) from None
    return Response(status_code=204)

Events

Every event shares the same envelope: type, timestamp (the moment the lifecycle edge happened, not the delivery time) and data. Event models use snake_case attributes for the camelCase payload keys - event.data.action_requests, request.decided_by_name, and so on. Event types newer than the installed SDK parse into UnknownEvent instead of raising, so handlers keep working across SDK versions.

The session events are how Live View talks back to your agent. SessionEndedEvent carries a reason (SessionEndReason.agent, timeout or manual), so you can tell a run that finished on its own from one a watcher disconnected or one that timed out. SessionActionEvent carries the action name a watcher pressed - a custom action you defined - and triggered_by, the member who pressed it; what the action means is up to your agent:

from vigilator_py_sdk import SessionActionEvent, SessionEndedEvent, SessionEndReason

@webhooks.on("session.action")
def on_action(event: SessionActionEvent) -> None:
    if event.data.action == "pause":
        pause_run(event.data.external_id)  # your correlation id from start_session

@webhooks.on("session.ended")
def on_ended(event: SessionEndedEvent) -> None:
    if event.data.reason is SessionEndReason.manual:
        stop_run(event.data.external_id)  # a watcher pulled the plug

Return a 2xx quickly and offload slow work - failed deliveries are retried on a backoff schedule. For local testing and delivery tooling, see the webhooks guide.

API Reference

Client

Client(
    key: str,
    base_url: str = "https://api.vigilator.dev",
    retries: int = 5,
    backoff_factor: float = 0.5,
    timeout: httpx.Timeout | float = 5.0,
)

create_interrupt(interrupt)

Creates an interrupt and returns it as an InterruptsPostResponse.

get_interrupt(interrupt_id)

Fetches a single interrupt by id as an InterruptsIdGetResponse, including any decisions taken on its action requests.

start_session(name, *, external_id=None, messages=None)

Starts a live session and returns it as a SessionsPostResponse.

append_session_messages(session_id, messages)

Appends messages to a running session and returns it as a SessionsIdMessagesPostResponse.

end_session(session_id)

Ends a session and returns it as a SessionsIdEndPostResponse. Ending an already-ended session is a no-op that returns the current state, so the call is safe to retry.

Input that breaks the API contract - an empty name, an empty message batch, more than 200 messages - raises pydantic.ValidationError before any request is sent.

close()

Closes the underlying HTTP connection pool. Called automatically when the client is used as a context manager.

WebhookHandler

WebhookHandler(secret: str, tolerance: float = 300.0)

on(event_type)

Decorator that registers a callback for one event type. Callbacks receive the typed event. Unknown types are allowed, so callbacks can target event types newer than the installed SDK.

handle(body, headers)

Verifies a delivery, parses it, invokes the callbacks registered for its type (synchronously, in registration order), and returns the typed event.

construct_event(body, headers)

Verifies a delivery and returns the typed event without dispatching - use this if you prefer routing events yourself.

verify(body, headers)

Verifies the signature only, without parsing the body.

All three verification methods raise WebhookVerificationError on missing or malformed Svix headers, a timestamp outside the tolerance window, a signature mismatch, or a verified body that is not valid JSON.

Errors

All SDK errors inherit from VigilatorError:

UsageLimitError through NotFoundError subclass APIError, so except APIError catches them all.