discli

┌──────────────┐    stdin (JSONL)     ┌──────────────────┐
│  Your Agent  │ ──────────────────▶  │                  │
│  (Python,    │                      │   discli serve   │──── Discord API
│   Node, etc) │ ◀──────────────────  │                  │
└──────────────┘    stdout (JSONL)    └──────────────────┘

{"event": "ready", "bot_id": "123456", "bot_name": "MyBot#1234"}
{"event": "message", "channel_id": "789", "content": "hello", "author": "alice", "author_id": "456", "message_id": "101112", "mentions_bot": true, "is_dm": false, ...}
{"event": "response", "req_id": "req-1", "ok": true, "message_id": "131415"}

{"action": "send", "channel_id": "444555666", "content": "Hello!", "req_id": "req-1"}

→ stdin:  {"action": "send", "channel_id": "444555666", "content": "Hello!", "req_id": "msg-1"}
← stdout: {"event": "response", "req_id": "msg-1", "ok": true, "message_id": "131415161718"}

import json
import subprocess
import threading
from queue import Queue


class ServeClient:
    def __init__(self, **kwargs):
        cmd = ["discli", "serve"]
        for key, value in kwargs.items():
            cmd.extend([f"--{key.replace('_', '-')}", str(value)])

        self.proc = subprocess.Popen(
            cmd,
            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=10)
            del self.pending[rid]
            return result

    def next_event(self):
        return self.events.get()


# Usage
client = ServeClient(events="messages")
ready = client.next_event()  # {"event": "ready", ...}

while True:
    event = client.next_event()
    if event["event"] == "message" and event.get("mentions_bot"):
        client.send({
            "action": "reply",
            "channel_id": event["channel_id"],
            "message_id": event["message_id"],
            "content": "Hello!",
        })

import asyncio
import json


class AsyncServeClient:
    def __init__(self):
        self.proc = None
        self.pending = {}
        self._counter = 0

    async def start(self, **kwargs):
        cmd = ["discli", "serve"]
        for key, value in kwargs.items():
            cmd.extend([f"--{key.replace('_', '-')}", str(value)])

        self.proc = await asyncio.create_subprocess_exec(
            *cmd,
            stdin=asyncio.subprocess.PIPE,
            stdout=asyncio.subprocess.PIPE,
        )

    async def send(self, action, wait=False):
        self._counter += 1
        rid = f"r{self._counter}"
        action["req_id"] = rid
        future = None
        if wait:
            future = asyncio.get_event_loop().create_future()
            self.pending[rid] = future
        self.proc.stdin.write((json.dumps(action) + "\n").encode())
        await self.proc.stdin.drain()
        if wait:
            return await asyncio.wait_for(future, timeout=10)

    async def read_events(self):
        while True:
            line = await self.proc.stdout.readline()
            if not line:
                break
            data = json.loads(line.decode().strip())
            rid = data.get("req_id")
            if rid in self.pending:
                self.pending.pop(rid).set_result(data)
            else:
                yield data


async def main():
    client = AsyncServeClient()
    await client.start(events="messages")

    async for event in client.read_events():
        if event["event"] == "ready":
            print(f"Ready as {event['bot_name']}")
        elif event["event"] == "message" and event.get("mentions_bot"):
            await client.send({
                "action": "reply",
                "channel_id": event["channel_id"],
                "message_id": event["message_id"],
                "content": "Hello!",
            })

asyncio.run(main())

← {"event": "response", "req_id": "req-1", "error": "Channel not found: 999"}

← {"event": "error", "message": "Invalid JSON: {bad json"}

result = client.send({"action": "send", "channel_id": "999", "content": "test"}, wait=True)
if "error" in result:
    print(f"Action failed: {result['error']}")
else:
    print(f"Sent message: {result['message_id']}")