Build a daily message digest

import Anthropic from '@anthropic-ai/sdk'
 
const anthropic = new Anthropic()
const API = 'https://api.bloomtext.com/v1'
 
const INSTRUCTIONS =
  'You summarize a clinician\'s BloomText messages for their morning digest. From the transcript, list every ' +
  'scheduled event, every follow-up someone is waiting on, and every to-do for the clinician. Report only what the ' +
  'messages say; never add clinical advice. Reply with JSON only: {"items": [{"kind": "event" | "follow_up" | "todo", ' +
  '"summary": string, "conversation_id": string, "due": string | null}]}'
 
async function listMessagesSince(accessToken, since) {
  const messages = []
  let cursor = null
  do {
    const url = new URL(`${API}/me/messages`)
    url.searchParams.set('after', since)
    url.searchParams.set('q', '-from:me')
    url.searchParams.set('group_by', 'conversation')
    url.searchParams.set('order', 'asc')
    url.searchParams.set('page[limit]', '100')
    if (cursor) url.searchParams.set('page[after]', cursor)
 
    const response = await fetch(url, { headers: { Authorization: `Bearer ${accessToken}` } })
    if (!response.ok) throw new Error(`BloomText API error: ${response.status}`)
    const { data, pagination } = await response.json()
    messages.push(...data)
    cursor = pagination.has_more ? pagination.next_cursor : null
  } while (cursor)
  return messages
}
 
function transcript(messages) {
  const byConversation = Object.groupBy(messages, (m) => m.conversation.id)
  return Object.values(byConversation).map((thread) => [
    `## ${thread[0].conversation.title} (conversation ${thread[0].conversation.id})`,
    ...thread.map((m) => `[${m.created_at}] ${m.sender.name} (${m.sender.kind}): ${m.body ?? `[file: ${m.file?.name}]`}`),
  ].join('\n')).join('\n\n')
}
 
// Run each morning for each connected clinician.
export async function buildDigest(clinician, { getAccessToken, saveDigest, markRun }) {
  const startedAt = new Date().toISOString()
  const accessToken = await getAccessToken(clinician) // refreshes, or returns null if they must reconnect
  if (!accessToken) return saveDigest(clinician, { reconnect: true })
 
  const messages = await listMessagesSince(accessToken, clinician.lastDigestAt)
  if (messages.length === 0) return markRun(clinician, startedAt)
 
  const reply = await anthropic.messages.create({
    model: 'claude-sonnet-5',
    max_tokens: 2000,
    system: INSTRUCTIONS,
    messages: [{ role: 'user', content: transcript(messages) }],
  })
  const text = reply.content[0].text
  const { items } = JSON.parse(text.slice(text.indexOf('{'), text.lastIndexOf('}') + 1)) // tolerate code fences
 
  await saveDigest(clinician, { items, messageCount: messages.length })
  await markRun(clinician, startedAt)
}

import json
from datetime import datetime, timezone
from itertools import groupby
 
import anthropic
import requests
 
client = anthropic.Anthropic()
API = "https://api.bloomtext.com/v1"
 
INSTRUCTIONS = (
    "You summarize a clinician's BloomText messages for their morning digest. From the transcript, list every "
    "scheduled event, every follow-up someone is waiting on, and every to-do for the clinician. Report only what the "
    "messages say; never add clinical advice. Reply with JSON only: "
    '{"items": [{"kind": "event" | "follow_up" | "todo", "summary": string, "conversation_id": string, "due": string | null}]}'
)
 
 
def list_messages_since(access_token: str, since: str) -> list[dict]:
    messages = []
    params = {"after": since, "q": "-from:me", "group_by": "conversation", "order": "asc", "page[limit]": 100}
    while True:
        response = requests.get(f"{API}/me/messages", headers={"Authorization": f"Bearer {access_token}"}, params=params)
        response.raise_for_status()
        page = response.json()
        messages.extend(page["data"])
        if not page["pagination"]["has_more"]:
            return messages
        params["page[after]"] = page["pagination"]["next_cursor"]
 
 
def transcript(messages: list[dict]) -> str:
    sections = []
    ordered = sorted(messages, key=lambda m: (m["conversation"]["id"], m["created_at"]))
    for conversation_id, thread in groupby(ordered, key=lambda m: m["conversation"]["id"]):
        thread = list(thread)
        section = [f"## {thread[0]['conversation']['title']} (conversation {conversation_id})"]
        for m in thread:
            text = m["body"] or f"[file: {(m['file'] or {}).get('name')}]"
            section.append(f"[{m['created_at']}] {m['sender']['name']} ({m['sender']['kind']}): {text}")
        sections.append("\n".join(section))
    return "\n\n".join(sections)
 
 
def build_digest(clinician, get_access_token, save_digest, mark_run):
    """Run each morning for each connected clinician."""
    started_at = datetime.now(timezone.utc).isoformat()
    access_token = get_access_token(clinician)  # refreshes, or returns None if they must reconnect
    if access_token is None:
        return save_digest(clinician, {"reconnect": True})
 
    messages = list_messages_since(access_token, clinician.last_digest_at)
    if not messages:
        return mark_run(clinician, started_at)
 
    reply = client.messages.create(
        model="claude-sonnet-5",
        max_tokens=2000,
        system=INSTRUCTIONS,
        messages=[{"role": "user", "content": transcript(messages)}],
    )
    text = reply.content[0].text
    items = json.loads(text[text.index("{"):text.rindex("}") + 1])["items"]  # tolerate code fences
 
    save_digest(clinician, {"items": items, "message_count": len(messages)})
    mark_run(clinician, started_at)