Hooks: Flow Control and Automation

Hooks are a powerful SDK feature that enable full workflow customization, including side-effects, intercepts, and redirects.

Suggest Edits

Overview

Hooks are a core SDK annotation that enable your flows to take arbitrarily complex behaviors. You can think of them as the mortar that binds all of your connected systems into a comprehensive workflow.

For example:

  • Special routing and auto-approve or -deny logic
  • Ensuring that only certain actors can approve or deny certain types of requests
  • Reflecting requests in external systems

Implementation

Hooks are all defined inside of a Sym Approval Flow's impl.py file, fire at specific points in the workflow, and can deliver a broad range of user-defined behaviors.

Hooks prefixed with on_ will fire after a relevant event has been triggered, but before it is processed. For example, an on_request Hook will trigger after a request form is submitted; but before the get_approvers Reducer is processed and a channel message is sent.

Hooks prefixed with after_ will fire after a relevant event has finished processing.

📘

Recursion is tightly gated

In theory, a Hook can return its own calling event (e.g. return ApprovalTemplate.request() inside of on_request). To prevent unintended timeouts due to user behavior, the Sym SDK will process the first recursive call only.

In other words, any recursive returns in Hooks should be written with the expectation that they will be processed exactly once, after which either a different outcome should be processed, or the Flow will terminate.

Determining if an event came from a Hook

You may want to split your logic based on the source of an event. For example, if you want to gate approvals to a specific list of people, but always allow SDK automations to pass through. In such cases, you'll want to inspect event.channel.type, as in the following example.

@hook
def on_approve(event):
    """
    Only let members of the approver safelist approve requests
    """
    # Don't check approval permissions if the event came from another hook
    if event.channel.type == "sdk":
        return

    if not has_approve_access(event):
        return ApprovalTemplate.ignore(
            message="You are not authorized to approve this request."
        )

For more information about event attributes, please see Working With Flow Fields and Data.

List of Hooks

HookFires whenUsage example
on_promptThe access request form is about to be shownRestricting access to certain flows
after_promptThe access request form has just been shownTracking usage in external systems
on_requestThe access request form is submittedAuto-approving requests which do not need outside approval
after_requestThe access request message has just been sentTracking requests in external systems
on_approveThe "approve" button on an access request message is clickedRestricting who may approve access to specific resources
after_approveThe access request message has just been updated to show an approvalTracking approvals in external systems
on_denyThe "deny" button on an access request message is clickedRestricting who many deny access to specific resources
after_denyThe access request message has just been updated to show a denialTracking denials in external systems
on_escalateA user is about to be given escalated permissionsTracking access grants in external systems
after_escalateA user has just been given escalated permissionsTracking access grants in external systems
on_deescalateA user is about to have their escalated permissions removedTracking access grant expirations in external systems
after_deescalateA user has just had their escalated permissions removedTracking access grant expirations in external systems

Examples

on_prompt

@hook
def on_prompt(event):
  if event.flow.srn.slug == "super_secret_flow" and event.user.email != "[email protected]":
    return ApprovalTemplate.ignore(message="You're not allowed to access the super secret Flow!")

after_prompt

@hook
def after_prompt(event):
  # Send a dad joke to everyone who wants to make an access request
  response = requests.get("icanhazdadjoke.com")
  print(response["joke"])

on_request

@hook
def on_request(event):
  # Auto-approve urgent requests for access by the person on call
  if event.payload.fields["urgency"] == "Urgent" and pagerduty.is_on_call(event.user):
    return ApprovalTemplate.approve()

after_request

@hook
def after_request(event):
  # Send a text to everyone on call when a request is made
  for user in pagerduty.users_on_call(escalation_policy_name="123"):
    requests.post(
      url="http://my-website.com/sms",
      body={"message": f"{event.user.email} just made an access request!"}
    )

on_approve

@hook
def on_approve(event):
  if not okta.is_user_in_group(event.user, group_id=event.flow.vars["okta_managers_group"]):
    return ApprovalTemplate.ignore(message="Only managers may approve access requests.")

after_approve

@hook
def after_approve(event):
  requester = event.get_actor("request")
  # Turn the requester's lamp green when they're approved
  requests.post(
    url="http://lamps.com",
    body={
      "color": "green",
      "lamp_owner": requester.email
    }
  )

on_deny

@hook
def on_deny(event):
  if okta.is_user_in_group(event.user, group_id=event.flow.vars["okta_interns_group"]):
    return ApprovalTemplate.ignore(message="Interns cannot deny access requests.")

after_deny

@hook
def after_deny(event):
  requester = event.get_actor("request")
  # Turn the requester's lamp red when they're denied
  requests.post(
    url="http://lamps.com",
    body={
      "color": "red",
      "lamp_owner": requester.email
    }
  )

on_escalate

@hook
def on_escalate(event):
  requester = event.get_actor("request")
  # Comment on the ticket tracking this access.
  requests.post(
    url=f"http://jira.com/comment-ticket", 
    body={
      "ticket_id": "123", 
      "message": f"{requester.email} has been granted access to {event.flow.srn.slug}."
    }
  )

after_escalate

@hook
def after_escalate(event):
  requester = event.get_actor("request")
  # Send a text to the requester when their access is granted
  requests.post(
    url="http://my-website.com/sms",
    body={
      "message": f"Your access to {event.flow.srn.slug} has been granted!",
      "user_email": requester.email
    }
  )

on_deescalate

@hook
def on_deescalate(event):
  # Comment on the ticket tracking this access.
  requests.post(
    url=f"http://jira.com/comment-ticket", 
    body={
      "ticket_id": "123", 
      "message": f"{event.user.email}'s {event.flow.srn.slug} access has ended."
    }
  )

after_deescalate

@hook
def after_escalate(event):
  # Send a text to the requester when their access has expired
  requests.post(
    url="http://my-website.com/sms",
    body={
      "message": f"Your access to {event.flow.srn.slug} has expired!",
      "user_email": event.user.email
    }
  )

Updated 10 days ago