本指南以五個階段逐步建立 Discord 代理，從約 20 行的關鍵字回應器，走到可直接上線的全自動化代理。每個階段建立在前一階段之上，所有範例皆為可執行完整程式。

五個階段

階段名稱行數功能
1反應型 Bot約 20回應關鍵字
2情境感知約 35回應前先抓取對話歷史
3主動代理約 60建立討論串、串流回應、意圖偵測
4多動作代理約 90單一事件下協調多個 Discord 行為
5全自主代理約 120含安全、防錯、率先降級與上線品質設計

第一階段：反應型 Bot

最簡化的代理。它監聽提及機器人的訊息，並以關鍵字比對回應。事件使用 discli listen 取得，回覆則用 discli message reply 傳送。

import json
import subprocess




def reply(channel_id, message_id, text):
    subprocess.run(
        ["discli", "message", "reply", channel_id, message_id, text],
        capture_output=True,
    )




proc = subprocess.Popen(
    ["discli", "--json", "listen", "--events", "messages"],
    stdout=subprocess.PIPE,
    text=True,
)


for line in proc.stdout:
    event = json.loads(line.strip())
    if not event.get("mentions_bot"):
        continue
    content = event["content"].lower()
    if "help" in content:
        reply(event["channel_id"], event["message_id"], "你好，我能幫你什麼？")
    elif "pricing" in content:
        reply(event["channel_id"], event["message_id"], "可參考 example.com/pricing")
    else:
        reply(event["channel_id"], event["message_id"], "已收到你的訊息，請稍等。")

為什麼可行

  • discli --json listen 會將 JSONL 事件逐行輸出到 stdout
  • 每筆事件都包含 mentions_bot、channel_id、message_id 與 content
  • discli message reply 是一次性指令，不需管理連線生命週期
  • for line in proc.stdout 會在每次事件到達前阻塞
Warning

可能的風險： 每次 reply() 都會啟一個新的 discli 行程，這會建立新的 Discord 連線、完成驗證與發訊息後斷線。每次回覆約需 2–3 秒。低流量 bot 可接受，流量高時不具擴展性。

第二階段：情境感知

第一階段每則訊息都獨立處理。情境感知代理會在回應前先抓最近的對話歷史，才能更準確理解使用者真正意圖。

import json
import subprocess




def reply(channel_id, message_id, text):
    subprocess.run(
        ["discli", "message", "reply", channel_id, message_id, text],
        capture_output=True,
    )




def get_context(channel_id, limit=5):
    """抓取近期訊息作為對話上下文。"""
    result = subprocess.run(
        ["discli", "--json", "message", "list", channel_id, "--limit", str(limit)],
        capture_output=True,
        text=True,
    )
    if result.returncode != 0:
        return []
    return json.loads(result.stdout)




def build_response(event, context):
    """使用對話上下文生成更精準回應。"""
    content = event["content"].lower()
    # 檢查使用者是否最近重複提問
    recent_questions = [m["content"] for m in context if m["author_id"] == event["author_id"]]
    if any("help" in q.lower() for q in recent_questions[:3]):
        return "看起來你最近已詢問過，先幫你轉接真人處理，請稍候。"
    if "help" in content:
        return "你好，我可以回答關於價格、安裝與帳務的問題。"
    if "pricing" in content:
        return "可參考 example.com/pricing，方案起跳價格為每月 10 美元。"
    return "收到你的訊息，能再補充一點你要的內容嗎？"




proc = subprocess.Popen(
    ["discli", "--json", "listen", "--events", "messages"],
    stdout=subprocess.PIPE,
    text=True,
)


for line in proc.stdout:
    event = json.loads(line.strip())
    if not event.get("mentions_bot"):
        continue
    context = get_context(event["channel_id"])
    response = build_response(event, context)
    reply(event["channel_id"], event["message_id"], response)

與第一階段差異

  • 新增 get_context()，回應前抓取最近 5 筆訊息
  • Bot 可以識別重複詢問並調整回覆
  • 回覆邏輯可參照歷史內容，而不只看當前這則訊息
Warning

可能的風險： 每次回應多了兩次 Discord 連線需求：一個是 message list，一個是 message reply。在高流量頻道中會增加延遲，且重啟後不保留上下文記憶。

第三階段：主動代理

在這個階段，架構會改變。第三階段改用 discli serve 建立持久雙向連線，不再是一次性 CLI 呼叫。代理可逐字串流回應、建立討論串，也可主動處理訊息而非等待提及。

import json
import subprocess
import sys
import threading




def start_serve():
    """以 subprocess 啟動 discli serve。"""
    return subprocess.Popen(
        ["discli", "serve", "--events", "messages"],
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
        text=True,
    )




def send_action(proc, 動作):
    """透過 stdin 將 JSONL 動作傳給 discli serve。"""
    proc.stdin.write(json.dumps(動作) + "\n")
    proc.stdin.flush()




def read_events(proc):
    """從 discli serve stdout 讀取 JSONL 事件。"""
    for line in proc.stdout:
        line = line.strip()
        if line:
            yield json.loads(line)




def detect_intent(content):
    """不用明確指令的簡易意圖判斷。"""
    content = content.lower()
    if any(w in content for w in ["help", "issue", "problem", "broken", "error"]):
        return "support"
    if any(w in content for w in ["how do i", "how to", "tutorial", "guide"]):
        return "question"
    return None




proc = start_serve()


# 等待就緒事件
for event in read_events(proc):
    if event.get("event") == "ready":
        print(f"機器人已就緒：{event['bot_name']}", file=sys.stderr)
        break


    # 主要事件迴圈
for event in read_events(proc):
    if event.get("event") != "message":
        continue
    if event.get("is_bot"):
        continue


    intent = detect_intent(event["content"])
    if intent is None:
        continue


    channel_id = event["channel_id"]
    message_id = event["message_id"]
    author = event["author"]


    if intent == "support":
        # 為使用者建立支援討論串
        send_action(proc, {
            "action": "thread_create",
            "channel_id": channel_id,
            "message_id": message_id,
            "name": f"支援：{author}",
            "content": f"{author}，我已建立討論串幫你處理，先簡述一下狀況。",
        })


    elif intent == "question":
        # 分段式串流回應
        send_action(proc, {"action": "typing_start", "channel_id": channel_id})
        response = "好問題！我先逐步說明..."


        # 啟動串流
        send_action(proc, {
            "action": "stream_start",
            "channel_id": channel_id,
            "reply_to": message_id,
            "req_id": "stream-1",
        })


        # 真正的代理會在這裡改接 LLM 串流輸出。
        # 示範中會一次送出整段回應。
        # 先等待 stream_start 回傳 stream_id，
        # 再陸續發送內容分段。參考 Streaming 文件取得更多細節。

與第二階段差異

  • 持久連線：discli serve 只維持一條 Discord WebSocket
  • 雙向 JSONL：透過 stdin 送入動作、從 stdout 接收事件
  • 建立討論串：可自動建立支援討論串，不需明確提及
  • 串流回應：可模擬即時輸入行為
  • 意圖偵測：不只依賴提及，能以關鍵模式回應
Warning

可能的風險： read_events 這個 generator 會阻塞主執行緒。若你發送會回傳結果的動作（如 stream_start），你必須讀取 stdout 取得 stream_id。若未使用 threading 或 async，容易進入死鎖；第四階段會處理此問題。

第四階段：多動作代理

實際上線的代理常會在單一事件中執行多個 Discord 行為。第四階段加入 threading，分離 stdin/stdout 讀寫，並可協調更複雜流程，例如新成員接待。

import json
import subprocess
import sys
import threading
import time
from collections import defaultdict
from queue import Queue




class DiscordAgent:
    def __init__(self):
        self.proc = subprocess.Popen(
            ["discli", "serve", "--events", "messages,members"],
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            text=True,
        )
        self.responses = Queue()
        self.events = Queue()
        self.pending = {}  # req_id -> 佇列
        self._req_counter = 0


        # 啟動 stdout 讀取執行緒
        self._reader = threading.Thread(target=self._read_loop, daemon=True)
        self._reader.start()


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


    def send(self, action, wait=False):
        """傳送動作。若 wait=True，會等待回應到達。"""
        self._req_counter += 1
        req_id = f"req-{self._req_counter}"
        action["req_id"] = req_id


        if wait:
            q = Queue()
            self.pending[req_id] = q


        self.proc.stdin.write(json.dumps(動作) + "\n")
        self.proc.stdin.flush()


        if wait:
            result = q.get(timeout=10)
            del self.pending[req_id]
            return result
        return None


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




# 設定
WELCOME_CHANNEL_ID = "1234567890"  # #welcome 頻道
MODS_CHANNEL_ID = "9876543210"     # #mod-log 頻道
NEWCOMER_ROLE_ID = "1111111111"    # @新進成員 身分組




def handle_member_join(agent, event):
    """完成新成員入群的完整流程。"""
    member_id = event["member_id"]
    member_name = event["member"]
    server_id = event["server_id"]


    # 1. 送出歡迎 DM
    agent.send({
        "action": "dm_send",
        "user_id": member_id,
        "content": (
            f"歡迎加入伺服器，{member_name}！👋\n\n"
            "可以照下面步驟開始：\n"
            "1. 在 #introductions 自我介紹\n"
            "2. 到 #roles 選擇身分組\n"
            "3. 到 #help 發問\n\n"
            "祝使用愉快！"
        ),
    })


    # 2. 指派 Newcomer 身分組
    agent.send({
        "action": "role_assign",
        "guild_id": server_id,
        "member_id": member_id,
        "role_id": NEWCOMER_ROLE_ID,
    })


    # 3. 在 #welcome 建立歡迎討論串
    result = agent.send({
        "action": "thread_create",
        "channel_id": WELCOME_CHANNEL_ID,
        "name": f"歡迎 {member_name}！",
        "content": f"大家好，{member_name} 剛剛加入，歡迎！🎉",
    }, wait=True)


    thread_id = result.get("thread_id")


    # 4. 通知管理員
    agent.send({
        "action": "send",
        "channel_id": MODS_CHANNEL_ID,
        "content": (
            f"📋 新成員加入：**{member_name}**（ID: {member_id}）\n"
            f"歡迎討論串：<#{thread_id}>" if thread_id else
            f"📋 新成員加入：**{member_name}**（ID: {member_id}）"
        ),
    })


    print(f"[onboard] 已完成 {member_name} 的入群流程", file=sys.stderr)




def handle_message(agent, event):
    """處理來訊，並回傳多個動作。"""
    if event.get("is_bot"):
        return
    content = event["content"].lower()
    channel_id = event["channel_id"]
    message_id = event["message_id"]


    if not event.get("mentions_bot"):
        return


    if "status" in content:
        # 先加表情回應，再分段回報狀態
        agent.send({
            "action": "reaction_add",
            "channel_id": channel_id,
            "message_id": message_id,
            "emoji": "⏳",
        })


        # 取得資料
        servers = agent.send({"action": "server_list"}, wait=True)
        server_count = len(servers.get("servers", []))


        agent.send({
            "action": "reply",
            "channel_id": channel_id,
            "message_id": message_id,
            "content": f"所有系統正常。正在監控 {server_count} 個伺服器。",
        })


        # 將沙漏表情改為打勾
        agent.send({
            "action": "reaction_remove",
            "channel_id": channel_id,
            "message_id": message_id,
            "emoji": "⏳",
        })
        agent.send({
            "action": "reaction_add",
            "channel_id": channel_id,
            "message_id": message_id,
            "emoji": "✅",
        })




# 主迴圈
agent = DiscordAgent()


# 等待就緒
while True:
    event = agent.next_event()
    if event.get("event") == "ready":
        print(f"機器人已就緒：{event['bot_name']}", file=sys.stderr)
        break


while True:
    event = agent.next_event()
    event_type = event.get("event")
    if event_type == "member_join":
        handle_member_join(agent, event)
    elif event_type == "message":
        handle_message(agent, event)

與第三階段差異

  • DiscordAgent 類別：封裝 serve 子程序，並正確處理 threading
  • 請求/回應關聯：用 req_id 將 wait=True 的回應對齊到原始請求
  • 多動作流程：handle_member_join 對一位新成員同時執行四個動作（DM、指派角色、建立討論串、發通知）
  • 非阻塞送出：未使用 wait=True 的動作直接丟出（即發即棄），可維持事件處理速度
Warning

可能的風險： send(wait=True) 中的 q.get(timeout=10) 可能因 Discord 回應慢或觸發速率限制而逾時；如果使用者關閉私訊，DM 也會失敗；硬編碼的頻道/角色 ID 在伺服器結構變動時會失效。第五階段會處理這些情況。

第五階段：全自主代理

包含權限設定檔、安全邊界、錯誤處理、速率限制感知、稽核並支援權限受限下的優雅降級，適合上線使用。

import json
import subprocess
import sys
import threading
import time
from queue import Queue, Empty




class DiscordAgent:
    """具備安全機制與錯誤處理的正式 Discord 代理。"""


    def __init__(self, profile="chat", slash_commands=None):
        cmd = [
            "discli", "serve",
            "--profile", profile,
            "--events", "messages,members,reactions",
        ]
        if slash_commands:
            cmd.extend(["--slash-commands", slash_commands])


        self.proc = subprocess.Popen(
            cmd,
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            text=True,
        )
        self.events = Queue()
        self.pending = {}
        self._req_counter = 0
        self._lock = threading.Lock()


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


    def _read_loop(self):
        try:
            for line in self.proc.stdout:
                line = line.strip()
                if not line:
                    continue
                try:
                    data = json.loads(line)
                except json.JSONDecodeError:
                    print(f"[warn] serve 回傳 JSON 格式錯誤：{line[:100]}", file=sys.stderr)
                    continue
                req_id = data.get("req_id")
                if req_id and req_id in self.pending:
                    self.pending[req_id].put(data)
                else:
                    self.events.put(data)
        except Exception as e:
            print(f"[fatal] 讀取執行緒發生錯誤：{e}", file=sys.stderr)
            self.events.put({"event": "error", "message": str(e)})


    def send(self, action, wait=False, timeout=15):
        """傳送動作，必要時等待回應並處理錯誤。"""
        with self._lock:
            self._req_counter += 1
            req_id = f"req-{self._req_counter}"
        action["req_id"] = req_id


        if wait:
            q = Queue()
            self.pending[req_id] = q


        try:
            self.proc.stdin.write(json.dumps(動作) + "\n")
            self.proc.stdin.flush()
        except (BrokenPipeError, OSError) as e:
            print(f"[fatal] 無法寫入 serve 行程：{e}", file=sys.stderr)
            return {"error": "serve process 已中止"}


        if wait:
            try:
                result = q.get(timeout=timeout)
            except Empty:
                result = {"error": "等待回應逾時"}
            finally:
                self.pending.pop(req_id, None)
            return result
        return {"ok": True}


    def safe_send(self, action, wait=False, fallback=None):
        """傳送動作，遇到錯誤則回傳 fallback。"""
        result = self.send(action, wait=wait)
        if "error" in result:
            print(f"[warn] 動作失敗：{action.get('action')} -> {result['error']}", file=sys.stderr)
            return fallback if fallback is not None else result
        return result


    def next_event(self, timeout=None):
        try:
            return self.events.get(timeout=timeout)
        except Empty:
            return None


    def shutdown(self):
        try:
            self.proc.stdin.close()
            self.proc.wait(timeout=5)
        except Exception:
            self.proc.kill()




class SupportAgent:
    """具備安全邊界的全自主支援代理。"""


    def __init__(self, config):
        self.config = config
        self.agent = DiscordAgent(
            profile=config.get("profile", "chat"),
            slash_commands=config.get("slash_commands"),
        )
        self.active_threads = {}  # user_id -> thread_id


    def run(self):
        """含錯誤復原的主要事件迴圈。"""
        # 等待 ready
        while True:
            event = self.agent.next_event(timeout=30)
            if event is None:
                print("[fatal] 等待 ready 逾時", file=sys.stderr)
                return
            if event.get("event") == "ready":
                print(f"[info] 機器人已就緒：{event['bot_name']}", file=sys.stderr)
                break
            if event.get("event") == "error":
                print(f"[error] 啟動錯誤：{event.get('message')}", file=sys.stderr)


        # 處理事件
        try:
            while True:
                event = self.agent.next_event(timeout=60)
                if event is None:
                    continue  # 心跳逾時，繼續等待
                try:
                    self._dispatch(event)
                except Exception as e:
                    print(f"[error] 事件處理器發生錯誤：{e}", file=sys.stderr)
        except KeyboardInterrupt:
            print("[info] 正在關閉…", file=sys.stderr)
        finally:
            self.agent.shutdown()


    def _dispatch(self, event):
        handlers = {
            "message": self._on_message,
            "member_join": self._on_member_join,
            "reaction_add": self._on_reaction,
            "slash_command": self._on_slash_command,
            "error": self._on_error,
        }
        handler = handlers.get(event.get("event"))
        if handler:
            handler(event)


    def _on_message(self, event):
        if event.get("is_bot"):
            return
        if not event.get("mentions_bot"):
            return


        channel_id = event["channel_id"]
        message_id = event["message_id"]
        user_id = event["author_id"]
        content = event["content"].lower()


        # 處理中顯示輸入中
        self.agent.send({"action": "typing_start", "channel_id": channel_id})


        try:
            # 檢查使用者是否已有進行中的支援討論串
            if user_id in self.active_threads:
                self.agent.safe_send({
                    "action": "reply",
                    "channel_id": channel_id,
                    "message_id": message_id,
                    "content": f"你目前已有在處理中的討論串：<#{self.active_threads[user_id]}>。",
                })
                return


            # 判斷意圖後回應
            if any(w in content for w in ["help", "issue", "problem", "bug"]):
                self._create_support_thread(event)
            elif any(w in content for w in ["thanks", "solved", "fixed"]):
                self.agent.safe_send({
                    "action": "reaction_add",
                    "channel_id": channel_id,
                    "message_id": message_id,
                    "emoji": "💚",
                })
            else:
                self._stream_response(event, "我可以怎麼幫你？提到問題後我可以幫你建立支援討論串。")
        finally:
            self.agent.send({"action": "typing_stop", "channel_id": channel_id})


    def _create_support_thread(self, event):
        """建立支援討論串並處理錯誤。"""
        result = self.agent.safe_send({
            "action": "thread_create",
            "channel_id": event["channel_id"],
            "message_id": event["message_id"],
            "name": f"支援：{event['author'][:50]}",
            "content": f"{event['author']}，我已建立這則討論串。\n\n請詳細描述你的問題。",
        }, wait=True, fallback=None)


        if result and result.get("thread_id"):
            self.active_threads[event["author_id"]] = result["thread_id"]
        else:
            # 降級處理：改以頻道回覆
            self.agent.safe_send({
                "action": "reply",
                "channel_id": event["channel_id"],
                "message_id": event["message_id"],
                "content": "討論串建立失敗，可能缺少權限。請直接在本頻道補充問題。",
            })


    def _stream_response(self, event, text):
        """進行串流回應並做完整清理。"""
        result = self.agent.send({
            "action": "stream_start",
            "channel_id": event["channel_id"],
            "reply_to": event["message_id"],
        }, wait=True)


        if "error" in result:
            # 回退為一般回覆
            self.agent.safe_send({
                "action": "reply",
                "channel_id": event["channel_id"],
                "message_id": event["message_id"],
                "content": text,
            })
            return


        stream_id = result["stream_id"]


        # 以詞組分段傳送（模擬 LLM 串流）
        words = text.split()
        for i in range(0, len(words), 3):
            chunk = " ".join(words[i:i+3]) + " "
            self.agent.send({
                "action": "stream_chunk",
                "stream_id": stream_id,
                "content": chunk,
            })
            time.sleep(0.1)


        self.agent.send({
            "action": "stream_end",
            "stream_id": stream_id,
        }, wait=True)


    def _on_member_join(self, event):
        """處理新成員加入，含降級策略。"""
        member_id = event["member_id"]
        member_name = event["member"]


        # 嘗試發送歡迎 DM（若 DM 關閉會失敗）
        self.agent.safe_send({
            "action": "dm_send",
            "user_id": member_id,
            "content": f"{member_name}，歡迎加入！有任何問題可到 #help 提問。",
        })


        # 若有設定歡迎頻道就同步公告
        welcome_channel = self.config.get("welcome_channel_id")
        if welcome_channel:
            self.agent.safe_send({
                "action": "send",
                "channel_id": welcome_channel,
                "content": f"歡迎 **{member_name}** 加入伺服器！",
            })


    def _on_reaction(self, event):
        """處理回應事件。"""
        # 例如：管理員按下勾號後關閉支援討論串
        if event.get("emoji") == "✅":
            # 若訊息在使用者討論串中，移除該追蹤
            for user_id, thread_id in list(self.active_threads.items()):
                if event.get("channel_id") == thread_id:
                    del self.active_threads[user_id]
                    self.agent.safe_send({
                        "action": "send",
                        "channel_id": thread_id,
                        "content": "此討論串已標示為已解決，感謝回報！",
                    })
                    break


    def _on_slash_command(self, event):
        """處理 slash command 與 followup 回覆。"""
        command = event.get("command")
        token = event.get("interaction_token")


        if command == "status":
            servers = self.agent.send({"action": "server_list"}, wait=True)
            count = len(servers.get("servers", []))
            active = len(self.active_threads)
            self.agent.safe_send({
                "action": "interaction_followup",
                "interaction_token": token,
                "content": f"線上中，監控 {count} 個伺服器，目前 {active} 個進行中支援討論串。",
            })
        else:
            self.agent.safe_send({
                "action": "interaction_followup",
                "interaction_token": token,
                "content": f"未識別指令：{command}",
            })


    def _on_error(self, event):
        print(f"[error] Discord 錯誤：{event.get('message')}", file=sys.stderr)




# ── Run ────────────────────────────────────────────────────────────
if __name__ == "__main__":
    config = {
        "profile": "chat",                       # 安全預設：避免破壞性動作
        "welcome_channel_id": "1234567890",       # 指定 #welcome 頻道
        "slash_commands": "commands.json",         # 選用 slash commands 檔
    }
    SupportAgent(config).run()

與第四階段差異

  • 權限設定檔：預設使用 chat，可降低誤做破壞性行為風險
  • safe_send()：將每個動作都包入錯誤處理並提供 fallback
  • 優雅降級：若討論串建立失敗（權限不足），改以頻道回覆補償
  • 結構化路由：用 _dispatch() 做事件分派，不再依賴過長的 if/elif
  • 執行緒安全：使用 lock 處理請求編號，並處理程序異常時的 broken pipe
  • 逾時處理：所有 wait=True 呼叫皆有逾時，不會無限等待
  • 稽核意識：chat 設定檔讓操作受控且完整紀錄
  • Slash command 支援：以 interaction_followup 回應 /status
  • 清理機制：shutdown() 會關閉 stdin 並等待程序結束
Warning

這個階段可能出錯的情況有限： 主要剩下：Discord API 異常（透過逾時緩解）、機器人權杖外洩（以環境變數替代硬編碼 token）、意圖判斷邏輯缺陷（透過日誌與監控補強）。

AI Serve Agent（ai_serve_agent.py）

在上述五個階段之外，discli 另外提供一個整合 Claude（透過 Anthropic Agent SDK）與 discli serve 的範例 AI 代理。這是建置智慧型 Discord 代理的建議架構。

架構

AI serve 代理採兩層設計：

  1. Claude Agent SDK 負責推理與決策
  2. discli serve 透過 JSONL 提供持續性 Discord 連線
  3. Bash 動作呼叫 讓 Claude 透過 bash 執行大部分 discli CLI 指令（發訊息、管理頻道、列出成員等）
  4. 元件區塊 讓 Claude 直接輸出豐富 UI 元件

Claude 從 discli serve 的 stdout 讀取事件並決定回應。一般行為走標準 CLI 指令；對互動元件（按鈕、選單、modals）則走元件區塊流程。

元件區塊

代理使用特殊 fenced code block 宣告訊息元件：

```component
{
  "channel_id": "444555666",
  "content": "請選擇一個操作：",
  "components": [
    {
      "type": "action_row",
      "components": [
        {"type": "button", "style": "primary", "label": "核可", "custom_id": "approve"},
        {"type": "button", "style": "danger", "label": "退回", "custom_id": "reject"}
      ]
    }
  ]
}
```

代理框架會解析這些區塊並轉成 JSONL 動作送到 discli serve。

Modal 需要兩段流程：先發送 modal_send 給使用者，然後處理 modal_submit 事件。代理維護 custom_id 對應的註冊表，當 modal_submit 來時即可對應到 handler。

範例測試訊息

可用以下範例訊息驗證能力：

  • @bot 幫我針對最愛語言發起投票：建立含按鈕的投票
  • @bot 顯示伺服器資訊：透過 Bash 執行 discli server info
  • @bot 建立回饋表單：開啟 modal 對話框
  • @bot 設定歡迎頻道：協調建立頻道並設定權限

執行範例

完整實作請見 agents/ai_serve_agent.py。此代理需安裝 anthropic 套件，並設定有效的 ANTHROPIC_API_KEY 環境變數。

選擇哪個階段

terminal

第一階段到第二階段

適用情境： 快速腳本、個人 bot、原型驗證。無需持久連線。每個動作為獨立 CLI 呼叫。

robot

第三階段到第五階段

適用情境： 上線級代理、即時互動、串流回應。透過 discli serve 維持單一持久連線。

架構決策：獨立 CLI 與 serve 對照

功能獨立 CLI（第一到二階段）discli serve（第三到五階段）
連線每個動作一次 HTTP 生命週期單一持久 Gateway 連線
延遲每次動作都含 HTTP 請求延遲通常低於 100ms / 動作
串流不支援全支援
討論串可建立（CLI）可建立並管理
輸入提示不實務完整支援
斜線指令不支援完整支援
複雜度最低中等

下一步

Last updated: August 2, 2026
7 min read
Edit this page