The Brew SDK provides methods to trigger automations and retrieve automation configuration. Automations are triggered via API with a typed payload that matches the automation’s configured schema.
New to automation triggers? Start with the Events & Automation Triggers guide to understand how triggers work, then return here for SDK-specific implementation details.
@dataclassclass TriggerResponse: success: bool execution_id: str # Unique execution ID for tracking automation_id: str # The triggered automation event_id: str # The event ID configured for this automation status: str # "running" contact: ContactResult # Contains email and action ("created" or "updated")@dataclassclass ContactResult: email: str # Contact email address action: str # "created" or "updated"
The contact field shows the result of the automatic contact upsert. When you trigger an automation, Brew automatically creates or updates the contact based on the email in your payload.
For detailed guides on common automation scenarios (welcome emails, order confirmations, payment recovery, trial reminders), see Events & Automation Triggers.
def trigger_welcome_flow(user): # Payload keys use camelCase to match the API client.automations.trigger( "auto_welcome_sequence", payload={ "email": user.email, "firstName": user.first_name, "plan": user.plan, "signupSource": user.referrer or "direct", "trialEndsAt": user.trial_end_date.isoformat() if user.trial_end_date else None, } )
Get the automation’s schema and validate locally before triggering:
Copy
Ask AI
info = client.automations.get_trigger_info("auto_abc123")schema = info.event_definition.payload_schema# Validate required fieldsrequired = schema.get("required", [])for field in required: if field not in payload: raise ValueError(f"Missing required field: {field}")
Handle errors gracefully
Always wrap trigger calls in try-except and handle specific error types:
Copy
Ask AI
from tenacity import retry, stop_after_attempt, wait_exponential@retry(stop=stop_after_attempt(3), wait=wait_exponential())def trigger_with_retry(automation_id, payload): try: return client.automations.trigger(automation_id, payload=payload) except BadRequestError: # Don't retry validation errors raise except RateLimitError as e: # Let tenacity handle the retry raise