Skip to main content

triggers.webhook

Webhook-based trigger support for durable agents.

Exposes HTTP endpoints that external systems (helpdesk, CI/CD, monitoring) can POST to in order to trigger an agent run. Source validation ensures only authorized origins can invoke the agent.

WebhookTriggerConfig Objects​

@dataclass
class WebhookTriggerConfig()

Configuration for a webhook-based trigger.

Attributes:

  • agent_id - The agent to trigger.
  • event_source - Allowed origin URL or hostname for source validation.
  • description - Human-readable description of the event.
  • enabled - Whether the trigger is active.
  • allowed_ips - Additional allowed IPs (optional).
  • secret - Shared secret for HMAC validation (optional, future).
  • metadata - Additional metadata.

WebhookTriggerResult Objects​

@dataclass
class WebhookTriggerResult()

Result of a webhook trigger execution.

WebhookTrigger Objects​

class WebhookTrigger()

Manages a webhook trigger configuration for an agent.

The actual HTTP handling is done by the FastAPI router endpoints. This class provides configuration storage and validation logic.

validate_source​

def validate_source(origin: str) -> bool

Validate that the incoming request is from an allowed source.

Checks:

  1. If event_source is empty, allow all (no restriction).
  2. If origin matches event_source hostname, allow.
  3. If origin is in allowed_ips, allow.

Arguments:

  • origin - The origin header or client IP of the request.

Returns:

True if the source is authorized.

register_webhook_trigger​

def register_webhook_trigger(config: WebhookTriggerConfig) -> WebhookTrigger

Register a webhook trigger for an agent.

unregister_webhook_trigger​

def unregister_webhook_trigger(agent_id: str) -> bool

Unregister a webhook trigger.

get_webhook_trigger​

def get_webhook_trigger(agent_id: str) -> WebhookTrigger | None

Get the webhook trigger config for an agent.

WebhookResponse Objects​

class WebhookResponse(BaseModel)

Response from a webhook trigger invocation.

WebhookInfo Objects​

class WebhookInfo(BaseModel)

Information about a webhook endpoint for external registration.

webhook_trigger​

@webhook_router.post("/{agent_id}/webhook", response_model=WebhookResponse)
async def webhook_trigger(agent_id: str, request: Request) -> WebhookResponse

Receive external events and trigger an agent run.

Event sources POST to this endpoint with a JSON payload. The agent's trigger config defines which event_source URLs are allowed.

Returns:

WebhookResponse with trigger status and optional workflow ID.

webhook_info​

@webhook_router.get("/{agent_id}/webhook", response_model=WebhookInfo)
async def webhook_info(agent_id: str) -> WebhookInfo

Return webhook URL and configuration for external systems to register.

This endpoint provides the information needed by external systems to configure their webhook callbacks.