Webhook

Webhook handling utilities for Arshai.

Classes

class arshai.web.webhook.WebhookConfig(secret=None, signature_header='X-Signature', signature_prefix='', hash_algorithm='sha256', max_body_size=10485760)

Bases: object

Configuration for webhook handler.

secret: Optional[str] = None

signature_header: str = 'X-Signature'

signature_prefix: str = ''

hash_algorithm: str = 'sha256'

max_body_size: int = 10485760

__init__(secret=None, signature_header='X-Signature', signature_prefix='', hash_algorithm='sha256', max_body_size=10485760)

class arshai.web.webhook.WebhookHandler(config=None)

Bases: object

Generic webhook handler with signature verification.

This is an OPTIONAL utility for handling webhooks. Existing Arshai applications are not affected by this addition.

Example

# Create handler with secret handler = WebhookHandler(WebhookConfig(secret=”my-secret”))

# In FastAPI/Flask route @app.post(“/webhook”) async def handle_webhook(request: Request):

# Verify and process data = await handler.process_request(request)

# Use with Arshai workflow workflow = MyWorkflow() result = await workflow.execute(IWorkflowState(data=data)) return result

Extending for specific platforms (in your application code):

class GitHubWebhookHandler(WebhookHandler):

def __init__(self, secret: str):

super().__init__(WebhookConfig(

secret=secret, signature_header=”X-Hub-Signature-256”, signature_prefix=”sha256=”, hash_algorithm=”sha256”

))

def is_pull_request_event(self, data: Dict[str, Any]) -> bool:

return “pull_request” in data

__init__(config=None)

Initialize webhook handler.

Parameters:

config (Optional[WebhookConfig]) – Optional configuration

verify_signature(body, signature)

Verify webhook signature.

Parameters:

• body (bytes) – Raw request body

• signature (str) – Signature from header

Return type:

bool

Returns:

True if signature is valid

async process_request(request)

Process incoming webhook request.

Parameters:

request (Any) – HTTP request object (FastAPI Request, Flask request, etc.)

Return type:

Dict[str, Any]

Returns:

Parsed webhook data

Raises:

WebhookValidationError – If validation fails

create_signature(body)

Create signature for outgoing webhooks.

Parameters:

body (Union[str, bytes, dict]) – Webhook body (string, bytes, or dict)

Return type:

str

Returns:

Signature string

class arshai.web.webhook.WebhookValidationError

Bases: Exception

Raised when webhook validation fails.