mirror/nanobot
1
0
Fork 0
mirror of https://github.com/HKUDS/nanobot.git synced 2026-08-04 16:38:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

docs: update channel plugin runtime event contract

Browse Source
This commit is contained in:
chengyongru 2026-07-01 16:55:24 +08:00 committed by Xubin Ren
parent 5f4cfbcb16
commit c757c5466c
1 changed files with 60 additions and 42 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

102
docs/channel-plugin-guide.md
View File

@ -103,7 +103,8 @@ class WebhookChannel(BaseChannel):
msg.content — markdown text (convert to platform format as needed) msg.content — markdown text (convert to platform format as needed)
msg.media — list of local file paths to attach msg.media — list of local file paths to attach
msg.chat_id — the recipient (same chat_id you passed to _handle_message) msg.chat_id — the recipient (same chat_id you passed to _handle_message)
msg.metadata — may contain "_progress": True for streaming chunks msg.metadata — channel routing context such as message/thread ids
msg.event — typed runtime event for progress/status messages
""" """
logger.info("[webhook] -> {}: {}", msg.chat_id, msg.content[:80]) logger.info("[webhook] -> {}: {}", msg.chat_id, msg.content[:80])
# In a real plugin: POST to a callback URL, send via SDK, etc. # In a real plugin: POST to a callback URL, send via SDK, etc.
@ -238,15 +239,15 @@ nanobot channels login <channel_name> --force # re-authenticate
| `supports_streaming` (property) | `True` when config has `"streaming": true` **and** subclass overrides `send_delta()`. | | `supports_streaming` (property) | `True` when config has `"streaming": true` **and** subclass overrides `send_delta()`. |
| `is_running` | Returns `self._running`. | | `is_running` | Returns `self._running`. |
| `login(force=False)` | Perform interactive login (e.g. QR code scan). Returns `True` if already authenticated or login succeeds. Override in subclasses that support interactive login. | | `login(force=False)` | Perform interactive login (e.g. QR code scan). Returns `True` if already authenticated or login succeeds. Override in subclasses that support interactive login. |
| `send_reasoning_delta(chat_id, delta, metadata?)` | Optional hook for streamed model reasoning/thinking content. Default is no-op. | | `send_reasoning_delta(chat_id, delta, metadata?, *, stream_id?)` | Optional hook for streamed model reasoning/thinking content. Default is no-op. |
| `send_reasoning_end(chat_id, metadata?)` | Optional hook marking the end of a reasoning block. Default is no-op. | | `send_reasoning_end(chat_id, metadata?, *, stream_id?)` | Optional hook marking the end of a reasoning block. Default is no-op. |
| `send_reasoning(msg)` | Optional one-shot reasoning fallback. Default translates to `send_reasoning_delta()` + `send_reasoning_end()`. | | `send_reasoning(msg)` | Optional one-shot reasoning fallback. Default translates to `send_reasoning_delta()` + `send_reasoning_end()`. |
### Optional (streaming) ### Optional (streaming)
| Method | Description | | Method | Description |
|--------|-------------| |--------|-------------|
| `async send_delta(chat_id, delta, metadata?)` | Override to receive streaming chunks. See [Streaming Support](#streaming-support) for details. | | `async send_delta(chat_id, delta, metadata?, *, stream_id?, stream_end=False, resuming=False)` | Override to receive streaming chunks. See [Streaming Support](#streaming-support) for details. |
### Message Types ### Message Types
@ -257,10 +258,12 @@ class OutboundMessage:
chat_id: str # recipient (same value you passed to _handle_message) chat_id: str # recipient (same value you passed to _handle_message)
content: str # markdown text — convert to platform format as needed content: str # markdown text — convert to platform format as needed
media: list[str] # local file paths to attach (images, audio, docs) media: list[str] # local file paths to attach (images, audio, docs)
metadata: dict # may contain: "_progress" (bool) for streaming chunks, metadata: dict # channel routing context, e.g. "message_id" for threading
# "message_id" for reply threading event: object | None # typed runtime/UI event; usually inspect with isinstance()
``` ```
Runtime/UI semantics live on `msg.event`. Plugin-authored outbound messages should not rely on legacy metadata flags such as `_progress`, `_stream_delta`, `_stream_end`, `_reasoning_delta`, `_turn_end`, or `_goal_status`; they are treated as ordinary metadata, not parsed as runtime events.
## Streaming Support ## Streaming Support
Channels can opt into real-time streaming — the agent sends content token-by-token instead of one final message. This is entirely optional; channels work fine without it. Channels can opt into real-time streaming — the agent sends content token-by-token instead of one final message. This is entirely optional; channels work fine without it.
@ -279,10 +282,18 @@ If either is missing, the agent falls back to the normal one-shot `send()` path.
Override `send_delta` to handle two types of calls: Override `send_delta` to handle two types of calls:
```python ```python
async def send_delta(self, chat_id: str, delta: str, metadata: dict[str, Any] | None = None) -> None: async def send_delta(
meta = metadata or {} self,
chat_id: str,
if meta.get("_stream_end"): delta: str,
metadata: dict[str, Any] | None = None,
*,
stream_id: str | None = None,
stream_end: bool = False,
resuming: bool = False,
) -> None:
buffer_key = stream_id or chat_id
if stream_end:
# Streaming finished — do final formatting, cleanup, etc. # Streaming finished — do final formatting, cleanup, etc.
return return
@ -290,12 +301,7 @@ async def send_delta(self, chat_id: str, delta: str, metadata: dict[str, Any] |
# delta contains a small chunk of text (a few tokens) # delta contains a small chunk of text (a few tokens)
``` ```
**Metadata flags:** Streaming state is passed through keyword-only arguments, not `_stream_delta` or `_stream_end` metadata flags. Use `stream_id` to key any per-stream buffers; fall back to `chat_id` when it is missing.
| Flag | Meaning |
|------|---------|
| `_stream_delta: True` | A content chunk (delta contains the new text) |
| `_stream_end: True` | Streaming finished (delta is empty) |
### Example: Webhook with Streaming ### Example: Webhook with Streaming
@ -310,18 +316,27 @@ class WebhookChannel(BaseChannel):
super().__init__(config, bus) super().__init__(config, bus)
self._buffers: dict[str, str] = {} self._buffers: dict[str, str] = {}
async def send_delta(self, chat_id: str, delta: str, metadata: dict[str, Any] | None = None) -> None: async def send_delta(
meta = metadata or {} self,
if meta.get("_stream_end"): chat_id: str,
text = self._buffers.pop(chat_id, "") delta: str,
metadata: dict[str, Any] | None = None,
*,
stream_id: str | None = None,
stream_end: bool = False,
resuming: bool = False,
) -> None:
buffer_key = stream_id or chat_id
if stream_end:
text = self._buffers.pop(buffer_key, "")
# Final delivery — format and send the complete message # Final delivery — format and send the complete message
await self._deliver(chat_id, text, final=True) await self._deliver(chat_id, text, final=True)
return return
self._buffers.setdefault(chat_id, "") self._buffers.setdefault(buffer_key, "")
self._buffers[chat_id] += delta self._buffers[buffer_key] += delta
# Incremental update — push partial text to the client # Incremental update — push partial text to the client
await self._deliver(chat_id, self._buffers[chat_id], final=False) await self._deliver(chat_id, self._buffers[buffer_key], final=False)
async def send(self, msg: OutboundMessage) -> None: async def send(self, msg: OutboundMessage) -> None:
# Non-streaming path — unchanged # Non-streaming path — unchanged
@ -350,7 +365,7 @@ When `streaming` is `false` (default) or omitted, only `send()` is called — no
| Method / Property | Description | | Method / Property | Description |
|-------------------|-------------| |-------------------|-------------|
| `async send_delta(chat_id, delta, metadata?)` | Override to handle streaming chunks. No-op by default. | | `async send_delta(chat_id, delta, metadata?, *, stream_id?, stream_end=False, resuming=False)` | Override to handle streaming chunks. No-op by default. |
| `supports_streaming` (property) | Returns `True` when config has `streaming: true` **and** subclass overrides `send_delta`. | | `supports_streaming` (property) | Returns `True` when config has `streaming: true` **and** subclass overrides `send_delta`. |
## Progress, Tool Hints, and Reasoning ## Progress, Tool Hints, and Reasoning
@ -359,18 +374,20 @@ Besides normal assistant text, nanobot can emit low-emphasis trace blocks. These
### Progress and Tool Hints ### Progress and Tool Hints
Progress and tool hints arrive through the normal `send(msg)` path. Check `msg.metadata` before rendering: Progress and tool hints arrive through the normal `send(msg)` path. Check `msg.event` before rendering:
```python ```python
async def send(self, msg: OutboundMessage) -> None: from nanobot.bus.outbound_events import ProgressEvent
meta = msg.metadata or {}
if meta.get("_tool_hint"): async def send(self, msg: OutboundMessage) -> None:
event = msg.event
if isinstance(event, ProgressEvent) and event.tool_hint:
# A short tool breadcrumb, e.g. read_file("config.json") # A short tool breadcrumb, e.g. read_file("config.json")
await self._send_trace(msg.chat_id, msg.content, kind="tool") await self._send_trace(msg.chat_id, msg.content, kind="tool")
return return
if meta.get("_progress"): if isinstance(event, ProgressEvent):
# Generic non-final status, e.g. "Thinking..." or "Running command..." # Generic non-final status, e.g. "Thinking..." or "Running command..."
await self._send_trace(msg.chat_id, msg.content, kind="progress") await self._send_trace(msg.chat_id, msg.content, kind="progress")
return return
@ -412,32 +429,33 @@ class WebhookChannel(BaseChannel):
chat_id: str, chat_id: str,
delta: str, delta: str,
metadata: dict[str, Any] | None = None, metadata: dict[str, Any] | None = None,
*,
stream_id: str | None = None,
) -> None: ) -> None:
meta = metadata or {} buffer_key = stream_id or chat_id
stream_id = str(meta.get("_stream_id") or chat_id) self._reasoning_buffers[buffer_key] = self._reasoning_buffers.get(buffer_key, "") + delta
self._reasoning_buffers[stream_id] = self._reasoning_buffers.get(stream_id, "") + delta await self._update_reasoning_block(chat_id, self._reasoning_buffers[buffer_key], final=False)
await self._update_reasoning_block(chat_id, self._reasoning_buffers[stream_id], final=False)
async def send_reasoning_end( async def send_reasoning_end(
self, self,
chat_id: str, chat_id: str,
metadata: dict[str, Any] | None = None, metadata: dict[str, Any] | None = None,
*,
stream_id: str | None = None,
) -> None: ) -> None:
meta = metadata or {} buffer_key = stream_id or chat_id
stream_id = str(meta.get("_stream_id") or chat_id) text = self._reasoning_buffers.pop(buffer_key, "")
text = self._reasoning_buffers.pop(stream_id, "")
if text: if text:
await self._update_reasoning_block(chat_id, text, final=True) await self._update_reasoning_block(chat_id, text, final=True)
``` ```
**Reasoning metadata flags:** **Reasoning arguments:**
| Flag | Meaning | | Argument | Meaning |
|------|---------| |------|---------|
| `_reasoning_delta: True` | A reasoning/thinking chunk; `delta` contains the new text. | | `delta` | A reasoning/thinking chunk for `send_reasoning_delta()`. |
| `_reasoning_end: True` | The current reasoning block is complete; `delta` is empty. | | `stream_id` | Stable id for this assistant turn/segment. Use it to key buffers instead of only `chat_id`. |
| `_reasoning: True` | Legacy one-shot reasoning. `BaseChannel.send_reasoning()` converts it to delta + end. | | `send_reasoning_end()` | The current reasoning block is complete. |
| `_stream_id` | Stable id for this assistant turn/segment. Use it to key buffers instead of only `chat_id`. |
Reasoning visibility is controlled by `showReasoning` globally or per channel: Reasoning visibility is controlled by `showReasoning` globally or per channel: