Project: Messaging Assistant

Capstone four: an agent living in Telegram or Slack — personalized through memory, guarded at the boundary, monitored like production. The front door of everything you've built.

▶ Watch this reel

What you'll learn

The bot surface
Tools & memory
Safety in a chat
Deploy & monitor

Remember this

Webhook receivers validate, enqueue, and answer in milliseconds; platform redeliveries dedupe on message IDs; placeholder-edit UX manages perceived latency
Memory personalizes: per-user namespaces, write gates at thread end, inspectable/erasable memory as a trust feature; group chats need per-sender trust ladders and mention-gating
Chat is the most exposed surface: untrusted inbound content, narrow egress, gated tools, per-user budgets — and reaction buttons close the quality feedback loop natively

Surface

Webhook: validate signature → filter (bots, non-text) → idempotent enqueue (message-ID keys) → 200 in ms.
Placeholder message + edit = chat's streaming UX; edits are rate-limited — batch.

Tools & memory

AG-12 registers chat-sized: compaction for long threads, per-user namespaces, end-of-thread write gates.
Group chats: mention-gating, per-sender trust ladder (known → suggest → basic), cross-user disclosure = incident.
User commands: 'what do you remember' / 'forget that' — memory as a trust feature.

Safety

Untrusted inbound content; narrow egress; inline-button confirmations; per-user daily budgets.

Deploy & monitor

Tiny always-on receiver + queue-driven workers; PY-23 pipeline with conversation golden set.
👍👎 reactions = native feedback loop; abuse-signal alerts; pilot with 5 friendly users, then widen.

Code: The chat worker: loop + memory + gates, compact

async def handle_chat(job: ChatJob):
    user = users.for_platform(job.user_id)
    trust = user.trust_level(job.chat_type)      # known/unknown/group

    ctx = context.build(
        system=CHAT_SYSTEM,                      # untrusted-content rules
        memory=memory.recall(job.text, user.id,  # threshold-gated
                             k=4),
        recent=recent_turns(job.chat_id, cap=8_000),
    )
    async with budget(user.daily_budget.remaining()):
        result = await agent_loop(ctx, tools=tools_for(trust))

    if result.needs_confirmation:                # inline buttons
        await send_buttons(job.chat_id, result.preview,
                           [Send, Edit, Cancel])
        return

    await telegram.edit(job.placeholder, result.reply)
    await react_buttons(job.chat_id)             # 👍👎 feedback
    memory.consider_write(user.id, job.thread)   # end-of-thread gate