cachibot-websocket-event

Add new WebSocket message types and event handlers to CachiBot's real-time streaming system. Use this skill when adding new real-time events, live updates, or bidirectional WebSocket communication — e.g., "add live progress events", "stream file upload status", "add approval request events".

仓库: jhd3197/CachiBot
最近来源活动: 2026年2月23日 22:27
检测到的 SKILL.md 语言: 英语
星标: 18
分支: 2

安装方式

默认使用会先检查来源的 Prompt；你也可以切换为直接命令，或下载本地副本。

检查来源文件

决定是否安装前，请先阅读 SKILL.md，以及 SkillsMP 当前展示的配套文件。

返回搜索返回安装

name	cachibot-websocket-event
description	Add new WebSocket message types and event handlers to CachiBot's real-time streaming system. Use this skill when adding new real-time events, live updates, or bidirectional WebSocket communication — e.g., "add live progress events", "stream file upload status", "add approval request events".
metadata	{"author":"cachibot","version":"1.0"}

CachiBot WebSocket Event Creation

Add new real-time WebSocket events spanning the backend server, frontend hook, and store integration.

Architecture Overview

Backend (Python)                    Frontend (TypeScript)
─────────────────                   ─────────────────────
websocket.py                        api/websocket.ts
  ├── ConnectionManager               └── WebSocketClient
  ├── WSMessage model                      ├── send(type, payload)
  └── run_agent() handler                  └── onMessage(handler)

models/websocket.py                 hooks/useWebSocket.ts
  ├── WSMessageType enum               └── message handler switch
  └── Payload models                       └── store actions

                                    stores/bots.ts
                                      └── state + actions

Step 1: Backend — Add Message Type

Edit cachibot/models/websocket.py:

class WSMessageType(str, Enum):
    # ... existing types ...
    YOUR_EVENT = "your_event"      # Server -> Client
    YOUR_REQUEST = "your_request"  # Client -> Server (if bidirectional)

class YourEventPayload(BaseModel):
    """Payload for your_event messages."""

    item_id: str
    status: str
    data: dict = {}

class WSMessage(BaseModel):
    type: WSMessageType
    payload: dict = {}

    # ... existing factory methods ...

    @classmethod
    def your_event(cls, item_id: str, status: str, data: dict | None = None) -> "WSMessage":
        return cls(
            type=WSMessageType.YOUR_EVENT,
            payload={"item_id": item_id, "status": status, "data": data or {}},
        )

from cachibot.api.websocket import manager

# Send to a specific client
await manager.send(client_id, WSMessage.your_event(
    item_id="abc",
    status="completed",
    data={"result": "..."},
))

# Or broadcast to all clients of a bot
await manager.broadcast(bot_id, WSMessage.your_event(...))

async def run_agent(client_id: str, ...):
    async for event in agent.run_stream(message):
        match event.event_type:
            # ... existing handlers ...
            case StreamEventType.your_event:
                await manager.send(
                    client_id,
                    WSMessage.your_event(event.data["item_id"], event.data["status"]),
                )

# In websocket.py, inside the message handling loop:
elif msg_type == WSMessageType.YOUR_REQUEST:
    payload = data.get("payload", {})
    # Process the request
    result = await process_your_request(payload)
    # Optionally send a response event back
    await manager.send(client_id, WSMessage.your_event(...))

// Add to WSMessageType (or wherever message types are defined)
export type WSMessageType =
  | 'chat'
  | 'cancel'
  | 'thinking'
  | 'tool_start'
  | 'tool_end'
  | 'message'
  | 'usage'
  | 'error'
  | 'done'
  | 'your_event'     // new
  | 'your_request'   // new (if client -> server)

// Add payload interface
export interface YourEventPayload {
  item_id: string
  status: string
  data: Record<string, unknown>
}

// Import store actions
import { useYourStore } from '../stores/your-store'

// Inside the message handler switch:
case 'your_event': {
  const payload = msg.payload as YourEventPayload
  // Update store with the event data
  useYourStore.getState().handleYourEvent(payload)
  break
}

// In api/websocket.ts — add to WebSocketClient class:
sendYourRequest(data: YourRequestPayload): void {
  this.send('your_request', data)
}

// In hooks/useWebSocket.ts — expose in the hook return:
const sendYourRequest = useCallback((data: YourRequestPayload) => {
  wsClient.sendYourRequest(data)
}, [])

return { ..., sendYourRequest }

// In your store:
interface YourState {
  // ... existing state ...
  handleYourEvent: (payload: YourEventPayload) => void
}

export const useYourStore = create<YourState>()(
  (set) => ({
    // ... existing state ...
    handleYourEvent: (payload) => set((state) => {
      // Update state based on the event
      return {
        items: state.items.map((item) =>
          item.id === payload.item_id
            ? { ...item, status: payload.status }
            : item
        ),
      }
    }),
  })
)

Type	Payload	Purpose
`thinking`	`{ content }`	LLM reasoning content
`tool_start`	`{ id, tool, args }`	Tool execution begins
`tool_end`	`{ id, result, success }`	Tool execution completes
`message`	`{ role, content, messageId, replyToId }`	Streaming text content
`usage`	`{ totalTokens, promptTokens, completionTokens, totalCost, ... }`	Token/cost stats
`error`	`{ message, code }`	Error occurred
`done`	`{ replyToId }`	Response complete
`platform_message`	`{ botId, chatId, role, content, ... }`	External platform message
`approval_needed`	`{ id, tool, args, risk_level, ... }`	Tool needs user approval