Support automation agents that resolve tickets on-behalf-of your users

ZENDESK_URL = (
    f"https://{ZENDESK_SUBDOMAIN}"
    f".zendesk.com/api/v2/tickets.json"
    f"?status=new"
)

def fetch_new_tickets() -> dict:
    resp = requests.get(
        ZENDESK_URL,
        auth=(ZENDESK_EMAIL + "/token",
              ZENDESK_API_TOKEN),
    )
    tickets = resp.json().get("tickets", [])
    new = [
        t for t in tickets
        if t["id"] not in ticket_thread_map
    ]
    return {"tickets": new}

def generate_suggested_reply(ticket: dict) -> dict:
    prompt = (
        f"Ticket: {ticket['subject']}\n"
        f"{ticket['description']}\n"
        "Write a helpful support reply."
    )
    try:
        events = _llm_runner.run(
            user_id=LLM_USER_ID,
            session_id=LLM_SESSION_ID,
            new_message=Content(
                role="user",
                parts=[Part(text=prompt)]
            ),
        )
        reply = "".join(e.text for e in events)
    except Exception:
        reply = "Thank you for reaching out! "\
                "We are looking into your issue."
    return {**ticket, "suggested_reply": reply}

def post_slack_digest(context: dict) -> dict:
    for ticket in context["tickets"]:
        tid = ticket["id"]
        if ticket_thread_map.get(str(tid)) == "digest":
            continue
        requests.post(
            "https://slack.com/api/chat.postMessage",
            headers={"Authorization":
                     f"Bearer {SLACK_BOT_TOKEN}"},
            json={
                "channel": SLACK_CHANNEL_ID,
                "text": (
                    f"Ticket #{tid}: "
                    f"{ticket['subject']}"
                ),
            },
        )
        ticket_thread_map[str(tid)] = "digest"
    save_ticket_thread_map(ticket_thread_map)
    return context

NOTION_URL = "https://api.notion.com/v1/pages"

def update_notion_kb(context: dict) -> dict:
    for ticket in context["tickets"]:
        tid = ticket["id"]
        key = f"notion_{tid}"
        if ticket.get("status") != "solved":
            continue
        if key in ticket_thread_map:
            continue
        requests.post(
            NOTION_URL,
            headers={
                "Authorization":
                    f"Bearer {NOTION_API_KEY}",
                "Notion-Version": "2022-06-28",
            },
            json={
                "parent": {"database_id":
                    NOTION_KB_DATABASE_ID},
                "properties": {
                    "Name": {"title": [{"text":
                        {"content":
                            f"Ticket {tid}"}}]},
                    "status": {"status":
                        {"name": ticket["status"]}},
                },
            },
        )
        ticket_thread_map[key] = True
    save_ticket_thread_map(ticket_thread_map)
    return context