One support agent for chat, email and phone

pip install noclick fastapi uvicorn
export NOCLICK_API_KEY=nk_live_...                # Full access, from the console's API keys
export APP_SECRET=$(python -c "import secrets; print(secrets.token_hex(32))")
export STRIPE_API_KEY=rk_test_...                 # a restricted key: see below
export SHOP_DOMAIN=yourshop.com                   # support@help.yourshop.com lives under it
export SUPPORT_TEAM_EMAIL=support-team@yourshop.com
export SUPPORT_TEAM_PHONE=+14155550142            # where handed-off calls ring
python 02_agents_in_your_product/support_agent/main.py

import base64
import hashlib
import hmac
import json
import os
import re
import sys
from dataclasses import dataclass
from pathlib import Path
from typing import Optional

import uvicorn
from fastapi import Depends, FastAPI, HTTPException, Request, Response

import noclick

app = FastAPI()

SHOP_DOMAIN = os.environ["SHOP_DOMAIN"]
SUPPORT_DOMAIN = f"help.{SHOP_DOMAIN}"   # the inbox's own subdomain: the shop's mail stays where it is
SUPPORT_TEAM = noclick.Human(
    email=os.environ["SUPPORT_TEAM_EMAIL"],   # refund approvals and handed-off chats and emails
    phone=os.environ["SUPPORT_TEAM_PHONE"],   # handed-off calls
)

HELP_DIR = Path(__file__).parent / "help"
EMBED_MODEL = "openrouter/openai/text-embedding-3-small"

MIGRATIONS = {
    "0001_help_chunks.sql": """
        create table help_chunks (
            id text primary key,
            title text not null,
            url text not null,
            body text not null,
            content_hash text not null,
            embedding halfvec(1536) not null,
            search tsvector generated always as (to_tsvector('english', title || ' ' || body)) stored
        );
        create index help_chunks_embedding on help_chunks using hnsw (embedding halfvec_cosine_ops);
        create index help_chunks_search on help_chunks using gin (search);
    """,
    "0002_search_help.sql": """
        create function search_help(question text, question_embedding halfvec, k int default 5)
        returns table (id text, title text, url text, body text, score float8)
        language sql stable as $$
            with q as materialized (
                -- any of the question's words, not all of them: questions are long
                select to_tsquery('simple', array_to_string(
                           tsvector_to_array(to_tsvector('english', question)), ' | ')) as ts
            ),
            semantic as (
                select id, row_number() over (order by embedding <=> question_embedding) as rank
                from help_chunks
                order by embedding <=> question_embedding
                limit 20
            ),
            keyword as (
                select id, row_number() over (order by ts_rank_cd(search, (select ts from q)) desc) as rank
                from help_chunks
                where search @@ (select ts from q)
                order by ts_rank_cd(search, (select ts from q)) desc
                limit 20
            )
            select c.id, c.title, c.url, c.body,
                   (coalesce(1.0 / (60 + s.rank), 0) + coalesce(1.0 / (60 + kw.rank), 0))::float8 as score
            from semantic s
            full join keyword kw using (id)
            join help_chunks c using (id)
            order by score desc
            limit k
        $$;
    """,
}

@dataclass
class Section:
    id: str
    title: str
    url: str
    body: str

    @property
    def content_hash(self) -> str:
        text = f"{EMBED_MODEL}\n{self.title}\n{self.url}\n{self.body}"
        return hashlib.sha256(text.encode()).hexdigest()


def read_help_center() -> list[Section]:
    sections = []
    for path in sorted(HELP_DIR.glob("*.md")):
        header, _, text = path.read_text().removeprefix("---\n").partition("\n---\n")
        article = dict(line.split(": ", 1) for line in header.splitlines())
        intro, *parts = text.split("\n## ")
        if intro.strip():
            sections.append(Section(path.stem, article["title"], article["url"], intro.strip()))
        for part in parts:
            heading, _, body = part.partition("\n")
            anchor = re.sub(r"[^a-z0-9]+", "-", heading.lower()).strip("-")
            title, url = f"{article['title']}: {heading}", f"{article['url']}#{anchor}"
            sections.append(Section(f"{path.stem}#{anchor}", title, url, body.strip()))
    return sections


UPSERT = """
    insert into help_chunks (id, title, url, body, content_hash, embedding)
    values (?, ?, ?, ?, ?, cast(? as halfvec))
    on conflict (id) do update set title = excluded.title, url = excluded.url, body = excluded.body,
        content_hash = excluded.content_hash, embedding = excluded.embedding
"""


def batches(items: list, size: int = 100) -> list:
    """A database `batch` runs up to 100 statements in one transaction."""
    return [items[start:start + size] for start in range(0, len(items), size)]


def index_help_center(help_center: noclick.Database) -> None:
    sections = read_help_center()
    rows = help_center.query("select id, content_hash from help_chunks").rows
    stored = {row["id"]: row["content_hash"] for row in rows}
    changed = [section for section in sections if stored.get(section.id) != section.content_hash]
    for batch in batches(changed):
        vectors = noclick.AI.embed([f"{s.title}\n\n{s.body}" for s in batch], model=EMBED_MODEL).vectors
        upserts = [(UPSERT, [s.id, s.title, s.url, s.body, s.content_hash, v]) for s, v in zip(batch, vectors)]
        help_center.batch(upserts)
    gone = sorted(set(stored) - {section.id for section in sections})
    for batch in batches(gone):
        help_center.batch([("delete from help_chunks where id = ?", [gone_id]) for gone_id in batch])
    print(f"Help center: {len(sections)} sections, {len(changed)} embedded, {len(gone)} removed.")

def stripe_connection() -> noclick.Connection:
    return noclick.Connection.from_name(
        "stripe",
        integration="stripe",
        credential_type="stripe_api_key",
        data={"api_key": os.environ["STRIPE_API_KEY"]},
        create_if_missing=True,
    )

REFUND_LIMIT_USD = 50


def hold_large_refunds() -> None:
    noclick.Policy.add(
        noclick.Rule(
            "stripe.create_refund",
            when=f"!has(args.amount) || args.amount > {REFUND_LIMIT_USD * 100}",
            approver=SUPPORT_TEAM,
        ),
        to="require_approval",
    )

CUSTOMER = "{{user.metadata.stripe_customer}}"

INSTRUCTIONS = f"""You are Northwind Coffee's support agent. Customers reach you in the chat
on our site, by email and by phone.

Answer from the help center, never from memory. Search it with help-center__query:
    select title, url, body from search_help(:question, noclick.embed(:question, '{EMBED_MODEL}'))
with the customer's question as `question`. If it doesn't cover something, say so and offer a person.

Orders: stripe__list_checkout_sessions lists the customer's orders with their items, and
stripe__list_charges their payments and refunds. Both return only the customer you're talking to.
If they aren't available, you don't know who the customer is: offer to hand them to our team, or
ask them to use the chat on our site while signed in, or to email us from the address on their
account.

Refunds: follow the refund policy in the help center. Always pass `amount`, in cents. A refund over
${REFUND_LIMIT_USD} waits for our team: tell the customer it's with the team and that you'll let
them know here.

When the customer asks for a person, is upset, or you can't help, call hand_off with a short summary.

When a customer emails, answer with your reply tool: that's how the email reaches them. In chat and
email, link the help article you used. On a call, speak in short sentences and don't read out
links. Never say something is done until a tool has done it."""


def support_agent(help_center: noclick.Database, stripe: noclick.Connection) -> noclick.Agent:
    return noclick.Agent.from_name(
        "support",
        create_if_missing=True,
        instructions=INSTRUCTIONS,
        tools=[
            help_center.only("query"),
            stripe.only("list_checkout_sessions").bind(
                customer=CUSTOMER, extra_params={"expand": ["data.line_items"]},
            ),
            stripe.only("list_charges").bind(customer=CUSTOMER),
            stripe.only("create_refund").bind(metadata={"requested_by": "{{user.id}}"}),
            SUPPORT_TEAM.only("hand_off"),
        ],
        budget_usd_per_month=300,
    ).hydrate()

CUSTOMER_CAP_USD = 2.0
SECRET = os.environ["APP_SECRET"].encode()


@dataclass
class Customer:
    id: str
    email: str
    name: str
    stripe_customer: Optional[str]


def signature(body: str) -> str:
    return hmac.new(SECRET, body.encode(), hashlib.sha256).hexdigest()


def signed_in(request: Request) -> Customer:
    body, _, signed = request.cookies.get("customer", "").rpartition(".")
    if not body or not hmac.compare_digest(signed, signature(body)):
        raise HTTPException(401, "Sign in first.")
    return Customer(**json.loads(base64.urlsafe_b64decode(body)))


@app.get("/api/demo/sign-in")
def demo_sign_in(email: str, name: str, stripe_customer: Optional[str] = None) -> Response:
    customer = {"id": email, "email": email, "name": name, "stripe_customer": stripe_customer}
    body = base64.urlsafe_b64encode(json.dumps(customer).encode()).decode()
    response = Response(f"Signed in as {name}. Open your site.")
    response.set_cookie("customer", f"{body}.{signature(body)}", httponly=True, samesite="lax")
    return response


def customer_user(customer: Customer) -> noclick.User:
    return noclick.User.from_id(
        customer.id,
        create_if_missing=True,
        name=customer.name,
        email=customer.email,
        metadata={"stripe_customer": customer.stripe_customer} if customer.stripe_customer else {},
        spend_cap_usd_per_month=CUSTOMER_CAP_USD,
    )


@app.get("/api/support/session")
def support_session(request: Request, customer: Customer = Depends(signed_in)) -> dict:
    channels = request.app.state.channels
    session = customer_user(customer).session(["agents"], ttl_s=3600)
    return {
        "session": session.token,
        "expires_at": session.expires_at.isoformat(),
        "user_id": customer.id,
        "agent_id": channels.agent.id,
        "email": channels.inbox.address,
        "phone": channels.number.number,
    }

@dataclass
class Channels:
    agent: noclick.Agent
    inbox: noclick.Inbox
    number: noclick.PhoneNumber


def support_inbox(support: noclick.Agent) -> noclick.Inbox:
    domain = noclick.Domain.connect(SHOP_DOMAIN)
    inbox = noclick.Inbox.from_name(
        "support", create_if_missing=True, local_part="support", domain=SUPPORT_DOMAIN, agent=support,
        senders="anyone",
    )
    if domain.refresh().status == "pending_dns":
        print(f"Mail to {inbox.address} needs these records in {SHOP_DOMAIN}'s DNS:")
        for record in domain.records:
            print(f"  {record.type} {record.name} {record.value}")
        if domain.setup_url:
            print(f"Or add them in one step: {domain.setup_url}")
    return inbox

AREA_CODE = "415"


def support_number(support: noclick.Agent) -> noclick.PhoneNumber:
    number = noclick.PhoneNumber.from_name("support", create_if_missing=True, area_code=AREA_CODE, max_monthly_usd=5,
                                           max_upfront_usd=5)
    return number.answer_with(support)

def set_up() -> Channels:
    help_center = noclick.Database.from_name("help-center", create_if_missing=True, engine="postgres")
    help_center.migrate(MIGRATIONS)
    index_help_center(help_center)
    hold_large_refunds()
    support = support_agent(help_center, stripe_connection())
    return Channels(agent=support, inbox=support_inbox(support), number=support_number(support))


def clean_up() -> None:
    noclick.PhoneNumber.from_name("support").release()
    noclick.Inbox.from_name("support").delete()
    noclick.Agent.from_name("support").delete()
    noclick.Database.from_name("help-center").delete()
    noclick.Policy.remove("stripe.create_refund")


def main() -> None:
    if sys.argv[1:] == ["clean-up"]:
        clean_up()
        return
    app.state.channels = set_up()
    channels = app.state.channels
    print(f"Support answers on {channels.inbox.address}, {channels.number.number} and the chat widget.")
    uvicorn.run(app, host="127.0.0.1", port=8000)


if __name__ == "__main__":
    main()