discli

Time:  0s      1.5s     3.0s     4.5s     5.0s
       │        │        │        │        │
Chunks: Hello,  I can    help     with     that!
Edits:  "..."   "Hello, I can "  "Hello, I can help with "  (stream_end) "Hello, I can help with that!"

import json
import subprocess
import threading
import time
from queue import Queue


class StreamingAgent:
    def __init__(self):
        self.proc = subprocess.Popen(
            ["discli", "serve", "--events", "messages"],
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            text=True,
        )
        self.events = Queue()
        self.pending = {}
        self._counter = 0

        threading.Thread(target=self._reader, daemon=True).start()

    def _reader(self):
        for line in self.proc.stdout:
            data = json.loads(line.strip())
            rid = data.get("req_id")
            if rid in self.pending:
                self.pending[rid].put(data)
            else:
                self.events.put(data)

    def send(self, action, wait=False):
        self._counter += 1
        rid = f"r{self._counter}"
        action["req_id"] = rid
        if wait:
            q = Queue()
            self.pending[rid] = q
        self.proc.stdin.write(json.dumps(action) + "\n")
        self.proc.stdin.flush()
        if wait:
            result = q.get(timeout=30)
            del self.pending[rid]
            return result
        return None

    def stream_to_discord(self, channel_id, reply_to, text_generator):
        """Stream text from a generator to a Discord message."""

        # Step 1: Start the stream
        self.send({"action": "typing_start", "channel_id": channel_id})
        result = self.send({
            "action": "stream_start",
            "channel_id": channel_id,
            "reply_to": reply_to,
        }, wait=True)
        self.send({"action": "typing_stop", "channel_id": channel_id})

        if "error" in result:
            # Fallback: collect all text and send as a regular message
            full_text = "".join(text_generator)
            self.send({
                "action": "reply",
                "channel_id": channel_id,
                "message_id": reply_to,
                "content": full_text,
            })
            return

        stream_id = result["stream_id"]

        # Step 2: Send chunks as they arrive
        try:
            for chunk in text_generator:
                self.send({
                    "action": "stream_chunk",
                    "stream_id": stream_id,
                    "content": chunk,
                })
        except Exception as e:
            # Send error message as final chunk
            self.send({
                "action": "stream_chunk",
                "stream_id": stream_id,
                "content": f"\n\n[Error: {e}]",
            })

        # Step 3: Finalize
        self.send({"action": "stream_end", "stream_id": stream_id}, wait=True)


def call_llm(prompt):
    """Call an OpenAI-compatible API and yield response chunks."""
    import openai

    client = openai.OpenAI()  # Uses OPENAI_API_KEY env var
    stream = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        stream=True,
    )
    for chunk in stream:
        delta = chunk.choices[0].delta.content
        if delta:
            yield delta


def main():
    agent = StreamingAgent()

    # Wait for ready
    while True:
        event = agent.events.get()
        if event.get("event") == "ready":
            print(f"Ready as {event['bot_name']}")
            break

    # Process messages
    while True:
        event = agent.events.get()
        if event.get("event") != "message":
            continue
        if not event.get("mentions_bot") or event.get("is_bot"):
            continue

        agent.stream_to_discord(
            channel_id=event["channel_id"],
            reply_to=event["message_id"],
            text_generator=call_llm(event["content"]),
        )


if __name__ == "__main__":
    main()

def handle_slash_command(agent, event):
    stream_result = agent.send({
        "action": "stream_start",
        "channel_id": event["channel_id"],
        "interaction_token": event["interaction_token"],
    }, wait=True)

    stream_id = stream_result["stream_id"]

    for chunk in call_llm(event["args"].get("question", "")):
        agent.send({
            "action": "stream_chunk",
            "stream_id": stream_id,
            "content": chunk,
        })

    agent.send({"action": "stream_end", "stream_id": stream_id})