Slack triage agents that act on-behalf-of your engineers

# user_mapping.json
# {
#   "U1234567890A": {
#     "scalekit_identifier": "user@example.com",
#     "slack_username": "Demo User",
#     "github_username": "demo-user"
#   }
# }

def get_user_identifier(self, slack_user_id):
    user_info = self.user_mappings.get(slack_user_id)
    if not user_info:
        print(f"No mapping found for Slack user {slack_user_id}")
        return None
    return user_info.get("scalekit_identifier")

params = {
    "channel": channel_id,
    "limit": limit,
    "oldest": str(oldest_time),   # per-channel checkpoint
}
if cursor:
    params["cursor"] = cursor

result = connector.execute_action_with_retry(
    identifier=identifier,
    tool="slack_fetch_conversation_history",
    connection_name=Settings.SCALEKIT_SLACK_CONNECTION,
    parameters=params,
)
messages = result.get("messages", [])

# Follow Slack's pagination to the end
cursor = result.get("response_metadata", {}).get("next_cursor")

workflow = StateGraph(MessageState)
workflow.add_node("analyze", self._analyze_message)
workflow.add_node("execute", self._execute_action)
workflow.add_node("confirm", self._send_confirmation)
workflow.set_entry_point("analyze")
workflow.add_edge("analyze", "execute")
workflow.add_edge("execute", "confirm")
workflow.add_edge("confirm", END)

def _analyze_with_rules(self, message):
    if any(k.lower() in message for k in Settings.GITHUB_KEYWORDS):
        return ActionType.GITHUB_ISSUE   # bug, issue, broken, error
    if any(k.lower() in message for k in Settings.ZENDESK_KEYWORDS):
        return ActionType.ZENDESK_TICKET # support, help, question
    return ActionType.IGNORE

result = self.connector.execute_action_with_retry(
    identifier=user_identifier,
    tool="github_issue_create",
    connection_name=Settings.SCALEKIT_GITHUB_CONNECTION,
    parameters={"owner": repo_owner, "repo": repo_name,
                "title": title, "body": body},
)
issue_number = result.get("number")
issue_url = result.get("html_url")

# Zendesk routing uses the same interface
result = self.connector.execute_action_with_retry(
    identifier=user_identifier,
    tool="zendesk_create_ticket",
    connection_name=Settings.SCALEKIT_ZENDESK_CONNECTION,
    parameters={"subject": subject, "description": description,
                "priority": priority, "type": ticket_type},
)

parameters = {
    "channel": channel_id,
    "text": text,   # issue number, URL, repository
}
if thread_ts:
    parameters["thread_ts"] = thread_ts  # reply in-thread

result = self.connector.execute_action_with_retry(
    identifier=user_identifier,
    tool="slack_send_message",
    connection_name=Settings.SCALEKIT_SLACK_CONNECTION,
    parameters=parameters,
)