Webhooks
Webhooks package for the Infrared SDK.
Provides webhook endpoint management (CRUD) and signature verification.
WEBHOOK_EVENT_FAILED
module-attribute
WEBHOOK_EVENT_FAILED: str = 'job.failed'
Event name sent when a job fails.
WEBHOOK_EVENT_RUNNING
module-attribute
WEBHOOK_EVENT_RUNNING: str = 'job.running'
Event name sent when a job starts running.
WEBHOOK_EVENT_SUCCEEDED
module-attribute
WEBHOOK_EVENT_SUCCEEDED: str = 'job.succeeded'
Event name sent when a job succeeds.
WebhooksServiceClient
Bases: ScrubbedSessionState
Client for managing webhook endpoints and verifying signatures.
Manages its own requests.Session and implements the context-manager
protocol for clean session teardown.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
api_key
|
str or _SharedApiKey
|
API key for authenticating requests. |
required |
logger
|
Logger
|
Logger instance for debug/info output. |
required |
base_url
|
str
|
Base URL for the Infrared API (e.g. |
required |
close
close() -> None
Close the underlying requests session.
register
register(url: str, type: str, ip: Optional[str] = None) -> WebhookEndpoint
Register a new webhook endpoint.
POST {base_url}/webhooks with the registration payload.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
url
|
str
|
The URL to receive webhook events. |
required |
type
|
str
|
Endpoint type ( |
required |
ip
|
str or None
|
Optional IP address filter. |
None
|
Returns:
| Type | Description |
|---|---|
WebhookEndpoint
|
The newly created endpoint (without timestamps). |
Raises:
| Type | Description |
|---|---|
WebhookRegistrationError
|
If the HTTP request fails. |
list
list() -> List[WebhookEndpoint]
List all webhook endpoints.
GET {base_url}/webhooks.
Returns:
| Type | Description |
|---|---|
list[WebhookEndpoint]
|
All registered endpoints (with timestamps). |
Raises:
| Type | Description |
|---|---|
WebhookError
|
If the HTTP request fails. |
get
get(endpoint_id: str) -> WebhookEndpoint
Get a specific webhook endpoint by ID.
GET {base_url}/webhooks/{endpoint_id}.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
endpoint_id
|
str
|
The ID of the endpoint to retrieve. |
required |
Returns:
| Type | Description |
|---|---|
WebhookEndpoint
|
The requested endpoint (with timestamps). |
Raises:
| Type | Description |
|---|---|
WebhookNotFoundError
|
If the endpoint is not found (404). |
WebhookError
|
If the HTTP request fails for other reasons. |
delete
delete(endpoint_id: str) -> None
Delete a webhook endpoint by ID.
DELETE {base_url}/webhooks/{endpoint_id}.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
endpoint_id
|
str
|
The ID of the endpoint to delete. |
required |
Raises:
| Type | Description |
|---|---|
WebhookNotFoundError
|
If the endpoint is not found (404). |
WebhookError
|
If the HTTP request fails for other reasons. |
verify_signature
staticmethod
verify_signature(
payload_body: bytes, headers: dict, secret: str, tolerance: int = 300
) -> bool
Verify a Standard Webhooks HMAC-SHA256 signature.
Validates that the signature in the webhook-signature header
matches the expected HMAC for the given payload, and that the
timestamp is within the tolerance window (replay protection).
Signing format:
signed_content = "{webhook-id}.{webhook-timestamp}.{body}"
signature = "v1," + base64(HMAC-SHA256(decoded_secret, signed_content))
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
payload_body
|
bytes
|
The raw request body bytes. |
required |
headers
|
dict
|
Request headers containing |
required |
secret
|
str
|
The webhook secret ( |
required |
tolerance
|
int
|
Maximum age of the timestamp in seconds (default 300). Set to 0 to disable replay protection. |
300
|
Returns:
| Type | Description |
|---|---|
bool
|
|
WebhookEndpoint
dataclass
Immutable snapshot of a webhook endpoint from the API.
The created_at and updated_at fields are optional because the
registration response (POST /webhooks) only returns {id, url, type}
without timestamps, whereas list/get responses include them.
Attributes:
| Name | Type | Description |
|---|---|---|
id |
str
|
Identifier of the endpoint. |
url |
str
|
The URL that receives webhook events. |
type |
str
|
Endpoint type ( |
created_at |
str or None
|
Creation time as returned by the API, or |
updated_at |
str or None
|
Last-update time as returned by the API, or |
from_response
classmethod
from_response(data: dict) -> WebhookEndpoint
Build a WebhookEndpoint from an API response dict.
Handles both registration responses (no timestamps) and list/get responses (with timestamps via camelCase keys).
WebhookError
Bases: Exception
Base exception for all webhook errors.
HTTP error context lives on attributes (status_code,
response_body) instead of the positional message so the raw
response body never lands in args[0] / default repr /
Sentry's message capture.
Attributes:
| Name | Type | Description |
|---|---|---|
status_code |
int or None
|
HTTP status of the failed response, if there was one. |
response_body |
str or None
|
The raw response text, if any. |
WebhookNotFoundError
WebhookRegistration
dataclass
Registration request payload for creating a webhook endpoint.
Attributes:
| Name | Type | Description |
|---|---|---|
url |
str
|
The URL to receive webhook events. |
type |
str
|
Endpoint type ( |
ip |
str or None
|
Optional IP address filter for the endpoint. |