Slack, Telegram & Google Chat

import asyncio
import os

from slack_bolt import App, Assistant
from slack_bolt.adapter.socket_mode import SocketModeHandler

from docsassistant.core import Citation, Done, Token  # see Channel-Agnostic Architecture

app = App(
    token=os.environ["SLACK_BOT_TOKEN"],                  # placeholder: xoxb-... bot token
    signing_secret=os.environ["SLACK_SIGNING_SECRET"],    # placeholder: from the app's Basic Information page
)
assistant = Assistant()
core = build_core()                                       # DocsAssistantCore, built once at startup


@assistant.thread_started
def start_thread(say, set_suggested_prompts):
    say("Hi, ask me anything about the documentation.")
    set_suggested_prompts(prompts=[
        {"title": "Install", "message": "How do I install the library?"},
        {"title": "Release notes", "message": "What changed in the latest release?"},
    ])


@assistant.user_message
def answer(payload, context, client, set_status, set_title):
    team_id, user_id = context.team_id, payload["user"]
    channel_id, thread_ts = payload["channel"], payload["thread_ts"]
    conversation_id = f"slack:{team_id}:{channel_id}:{thread_ts}"

    set_status(status="is searching the docs...")
    streamer = client.chat_stream(
        channel=channel_id,
        thread_ts=thread_ts,
        recipient_team_id=team_id,
        recipient_user_id=user_id,
    )

    async def run() -> list[Citation]:
        sources: list[Citation] = []
        async for event in core.ask(conversation_id, f"slack:{team_id}:{user_id}", payload["text"]):
            if isinstance(event, Token):
                streamer.append(markdown_text=event.text)
            elif isinstance(event, Citation):
                sources.append(event)
            elif isinstance(event, Done):
                break
        return sources

    sources = asyncio.run(run())                         # Bolt's sync listeners run in pool threads with no running loop
    footer = "\n\n" + "\n".join(f"[{c.marker}] <{c.url}|{c.title}>" for c in sources)
    streamer.append(markdown_text=footer)
    streamer.stop(blocks=[feedback_block()])             # thumbs up / down, see below


app.use(assistant)

if __name__ == "__main__":
    SocketModeHandler(app, os.environ["SLACK_APP_TOKEN"]).start()   # placeholder: xapp-... app-level token

import { App } from "@slack/bolt";

const app = new App({
  token: process.env.SLACK_BOT_TOKEN,               // placeholder: xoxb-... bot token
  signingSecret: process.env.SLACK_SIGNING_SECRET,  // placeholder
});

app.message(async ({ message, sayStream, setStatus }) => {
  if (!("text" in message) || message.subtype) return;

  await setStatus({ status: "Searching the docs..." });
  const streamer = sayStream();

  // docsAssistant.ask is a thin client for the shared core, for example the /v1/ask SSE API
  for await (const event of docsAssistant.ask(message.thread_ts ?? message.ts, message.user, message.text)) {
    if (event.type === "token") {
      await streamer.append({ markdown_text: event.text });
    }
  }
  await streamer.stop();
});

await app.start(Number(process.env.PORT ?? 3000));

import com.slack.api.bolt.App;
import com.slack.api.bolt.AppConfig;
import com.slack.api.bolt.middleware.builtin.Assistant;
import com.slack.api.model.assistant.SuggestedPrompt;

import java.util.Collections;

AppConfig config = AppConfig.builder()
        .singleTeamBotToken(System.getenv("SLACK_BOT_TOKEN"))        // placeholder: xoxb-... bot token
        .signingSecret(System.getenv("SLACK_SIGNING_SECRET"))        // placeholder
        .build();
App app = new App(config);

Assistant assistant = new Assistant(app.executorService());

assistant.threadStarted((req, ctx) -> {
    try {
        ctx.say("Hi, ask me anything about the documentation.");
        ctx.setSuggestedPrompts(Collections.singletonList(
                SuggestedPrompt.create("How do I install the library?")));
    } catch (Exception e) {
        ctx.logger.error("Failed to handle assistant thread started event", e);
    }
});

assistant.userMessage((req, ctx) -> {
    try {
        ctx.setStatus("is searching the docs...");
        var event = req.getEvent();
        String conversationId = "slack:" + ctx.getTeamId() + ":" + event.getChannel() + ":" + event.getThreadTs();
        String answer = core.ask(conversationId, "slack:" + ctx.getTeamId() + ":" + event.getUser(), event.getText())
                .collectList()
                .map(Slack::renderMarkdown)           // join tokens, append the source links
                .block();
        ctx.say(answer);
    } catch (Exception e) {
        ctx.logger.error("Failed to answer", e);
    }
});

app.assistant(assistant);

import os
import time

from telegram import Update
from telegram.ext import Application, ContextTypes, MessageHandler, filters

from docsassistant.core import Citation, Done, Token

UPDATE_INTERVAL = 0.5  # seconds between draft updates; tune against Telegram's rate limits


async def on_text(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    chat = update.effective_chat
    user = update.effective_user
    core = context.application.bot_data["core"]
    draft_id = update.update_id or 1                     # any non-zero identifier unique to this answer

    await context.bot.send_message_draft(chat_id=chat.id, draft_id=draft_id)   # empty text = thinking placeholder

    text, sources, last_push = "", [], 0.0
    async for event in core.ask(f"telegram:{chat.id}", f"telegram:{user.id}", update.effective_message.text):
        if isinstance(event, Token):
            text += event.text
            if time.monotonic() - last_push >= UPDATE_INTERVAL:
                await context.bot.send_message_draft(chat_id=chat.id, draft_id=draft_id, text=text[:4096])
                last_push = time.monotonic()
        elif isinstance(event, Citation):
            sources.append(event)
        elif isinstance(event, Done):
            break

    links = "\n".join(f"[{c.marker}] {c.title}: {c.url}" for c in sources)
    await context.bot.send_message(chat_id=chat.id, text=(text + "\n\n" + links)[:4096])   # persists the answer


def main() -> None:
    application = (Application.builder()
                   .token(os.environ["TELEGRAM_BOT_TOKEN"])   # placeholder: from BotFather
                   .concurrent_updates(True)                  # one chat's long answer must not block the others
                   .build())
    application.bot_data["core"] = build_core()
    application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, on_text))
    application.run_polling()


if __name__ == "__main__":
    main()

import org.telegram.telegrambots.client.okhttp.OkHttpTelegramClient;
import org.telegram.telegrambots.longpolling.TelegramBotsLongPollingApplication;
import org.telegram.telegrambots.longpolling.util.LongPollingSingleThreadUpdateConsumer;
import org.telegram.telegrambots.meta.api.methods.send.SendMessage;
import org.telegram.telegrambots.meta.api.methods.send.SendMessageDraft;
import org.telegram.telegrambots.meta.api.objects.Update;
import org.telegram.telegrambots.meta.generics.TelegramClient;

public class DocsTelegramBot implements LongPollingSingleThreadUpdateConsumer {

    private final TelegramClient client;
    private final AssistantCore core;

    DocsTelegramBot(String token, AssistantCore core) {
        this.client = new OkHttpTelegramClient(token);
        this.core = core;
    }

    @Override
    public void consume(Update update) {
        if (update.getMessage() == null || !update.getMessage().hasText()) {
            return;
        }
        long chatId = update.getMessage().getChatId();
        long userId = update.getMessage().getFrom().getId();
        int draftId = update.getUpdateId();                          // any non-zero id unique to this answer

        var text = new StringBuilder();
        core.ask("telegram:" + chatId, "telegram:" + userId, update.getMessage().getText())
                .filter(e -> e instanceof CoreEvent.Token)
                .sample(java.time.Duration.ofMillis(500))            // throttle drafts
                .doOnNext(e -> {
                    text.setLength(0);
                    text.append(Telegram.joinTokensSoFar(e));        // hypothetical helper
                    send(SendMessageDraft.builder().chatId(chatId).draftId(draftId).text(text.toString()).build());
                })
                .blockLast();

        send(SendMessage.builder().chatId(chatId).text(text.toString()).build());
    }

    private void send(org.telegram.telegrambots.meta.api.methods.botapimethods.BotApiMethod<?> method) {
        try {
            client.execute(method);
        } catch (Exception e) {
            throw new IllegalStateException(e);
        }
    }

    public static void main(String[] args) throws Exception {
        String token = System.getenv("TELEGRAM_BOT_TOKEN");           // placeholder: from BotFather
        try (var app = new TelegramBotsLongPollingApplication()) {
            app.registerBot(token, new DocsTelegramBot(token, buildCore()));
            Thread.currentThread().join();
        }
    }
}

import os

from fastapi import BackgroundTasks, FastAPI, Request
from google.oauth2 import service_account
from googleapiclient.discovery import build

from docsassistant.core import Citation, Done, Token

app = FastAPI()
core = build_core()
credentials = service_account.Credentials.from_service_account_file(
    os.environ["GOOGLE_APPLICATION_CREDENTIALS"],        # placeholder: path to the app's service-account key file
    scopes=["https://www.googleapis.com/auth/chat.bot"],
)
chat_api = build("chat", "v1", credentials=credentials)


@app.post("/google-chat")
async def receive(request: Request, background: BackgroundTasks):
    verify_google_bearer_token(request)                   # see the Chat app authentication docs
    event = await request.json()
    payload = event["chat"]["messagePayload"]
    message = payload["message"]

    background.add_task(
        answer,
        space=payload["space"]["name"],
        thread=message["thread"]["name"],
        user=event["chat"]["user"]["name"],
        text=message.get("argumentText") or message["text"],
    )
    return {}                                             # empty acknowledgement within 30 seconds


async def answer(space: str, thread: str, user: str, text: str) -> None:
    body, sources = "", []
    async for ev in core.ask(f"gchat:{space}:{thread}", f"gchat:{user}", text):
        if isinstance(ev, Token):
            body += ev.text
        elif isinstance(ev, Citation):
            sources.append(ev)
    links = "\n".join(f"[{c.marker}] <{c.url}|{c.title}>" for c in sources)
    chat_api.spaces().messages().create(
        parent=space,
        body={"text": f"{body}\n\n{links}", "thread": {"name": thread}},
        messageReplyOption="REPLY_MESSAGE_FALLBACK_TO_NEW_THREAD",
    ).execute()