BotAPI 协议

源码路径: nahida_bot_sdk.api

BotAPI protocol, ChannelService protocol, and related interfaces.

---

类

LLMUsage

Token usage from an LLM call.

字段	类型	默认值	描述
input_tokens	int	0
output_tokens	int	0
cached_tokens	int	0
reasoning_tokens	int	0

LLMResponse

Normalized response from a single-turn LLM chat call.

字段	类型	默认值	描述
content	str	—
model	str	''
provider	str	''
finish_reason	str	''
usage	`LLMUsage	None`	None
tool_calls	list[dict[str, Any]]	[]

SubagentResult

Result from a multi-turn subagent run.

字段	类型	默认值	描述
final_response	str	—
status	str	'succeeded'
model	str	''
provider	str	''
steps	int	0
usage	`LLMUsage	None`	None
error	str	''

PluginLogger

Structured logger automatically scoped to a plugin.

• 基类: Protocol

方法:

debug(msg: str, kwargs: object = {})

info(msg: str, kwargs: object = {})

warning(msg: str, kwargs: object = {})

error(msg: str, kwargs: object = {})

exception(msg: str, kwargs: object = {})

SubscriptionHandle

Handle returned by subscribe(); call unsubscribe() to detach.

• 基类: Protocol

方法:

unsubscribe()

WebhookHandle

Handle returned by register_webhook_endpoint().

• 基类: Protocol

方法:

unsubscribe()

WebhookRequest

Raw HTTP request delivered to a plugin-owned webhook endpoint.

字段	类型	默认值	描述
method	str	—
path	str	—
headers	dict[str, str]	—
query	dict[str, str]	—
body	bytes	—
client_host	str	''

WebhookResponse

Raw HTTP response returned by a plugin-owned webhook endpoint.

字段	类型	默认值	描述
status_code	int	204
body	`bytes	str`	b''
headers	dict[str, str]	{}

ManagedTempFile

A plugin-owned temporary file path managed by the bot runtime.

字段	类型	默认值	描述
path	str	—
plugin_id	str	''
cleanup_token	str	''
ttl_seconds	int	3600

方法:

as_attachment(type: str, filename: str = '', mime_type: str = '', caption: str = '', cleanup_after_send: bool = True)

Create an outbound attachment bound to this managed temporary file.

ChannelService

Runtime contract for a channel service exposed by a plugin.

Channel services are ordinary plugins that explicitly register themselves with BotAPI.register_channel().

• 基类: Protocol

属性 (Properties):

channel_id

• 返回类型: str

Unique channel/platform identifier.

方法:

handle_inbound_event(event: dict[str, Any])

Normalize one platform-native event and publish a bot event.

send_message(target: str, message: OutboundMessage)

Send one normalized outbound message to the channel.

BotAPI

Interface that plugins use to interact with the bot runtime.

The concrete implementation is injected at load time; tests inject a mock.

• 基类: Protocol

属性 (Properties):

scheduler_service

• 返回类型: Any | None

Scheduler service exposed to plugins that provide scheduler tools.

logger

• 返回类型: PluginLogger

Structured logger scoped to this plugin.

方法:

send_message(target: str, message: OutboundMessage, channel: str = '')

Send a message to an external target. Returns platform message ID.

record_session_event(session_id: str, content: str, source: str = '', metadata: dict[str, Any] | None = None)

Write a system turn into a session's history without triggering a run.

request_agent_response(message: InboundMessage, session_id: str = '', reason: str = '', instruction: str = '', observed_messages: tuple[InboundMessage, ...] = (), reply_to_message_id: str | None = None)

Ask the main router to run the agent for a group conversation.

record_message_delivery(target: ChatAddress | str, text: str, source: str, delivery_mode: str = '', status: str = 'sent', message_id: str = '', error: str = '', metadata: dict[str, Any] | None = None, source_session_id: str = '', source_chat_address: str = '', source_user_id: str = '')

Write an outbound delivery audit record without affecting memory.

create_temp_file(suffix: str = '', prefix: str = '', purpose: str = '', ttl_seconds: int = 3600)

Allocate a plugin-scoped temporary file managed by the bot runtime.

cleanup_temp_files(expired_only: bool = True)

Clean this plugin's managed temporary files.

cleanup_temp_attachment(attachment: Attachment)

Clean one managed temporary attachment.

on_event(event_type: type)

Decorator: register an event handler.

subscribe(event_type: type, handler: Callable[..., Awaitable[None]])

Programmatic event subscription. Returns an unsubscribe handle.

register_tool(name: str, description: str, parameters: dict[str, Any], handler: Callable[..., Awaitable[str]])

Register a tool that the LLM can call during conversations.

unregister_tool(name: str)

Remove a previously registered tool by name. Returns True if found.

register_channel(channel: ChannelService)

Register a channel service implemented by this plugin.

register_provider_type(type_key: str, factory: Callable[[dict[str, Any]], Any], config_schema: dict[str, Any] | None = None, description: str = '')

Register a provider type that can be used from YAML config.

register_webhook_endpoint(path: str, handler: Callable[[WebhookRequest], Awaitable[WebhookResponse | None]], methods: tuple[str, ...] = ('POST',))

Register a plugin-owned raw HTTP webhook endpoint.

register_prompt_supplement(key: str, instruction: str, channel: str | None = None, filter: Callable[[MessageContext], bool] | None = None)

Register a supplemental instruction to inject into the system prompt.

The supplement is appended after the base prompt and behavioral instructions. When channel is provided, the supplement is only injected for messages originating from that channel. When filter is provided, it is called with the current MessageContext. When both are given, both must match (AND logic). When neither is provided, the supplement is always injected.

Args: key: Unique identifier for this supplement within the plugin. instruction: The prompt text to inject. channel: Optional channel name to restrict injection to. filter: Optional callable for complex matching conditions.

unregister_prompt_supplement(key: str)

Remove a previously registered prompt supplement. Returns True if found.

register_status_provider(key: str, handler: Callable[..., Awaitable[str | None]], label: str = '')

Register a provider that contributes text to /status output.

The handler is an async callable invoked with keyword arguments session_id and chat_key. It should return a short text block describing the plugin's state for that chat, or None to contribute nothing.

unregister_status_provider(key: str)

Remove a previously registered status provider. Returns True if found.

collect_status_providers(session_id: str, chat_key: str)

Collect text blocks from all registered status providers.

register_command(name: str, handler: Callable[..., Awaitable[CommandHandlerResult]], description: str = '', aliases: list[str] | None = None)

Register a /command that is matched from incoming messages.

get_session(session_id: str)

Look up session metadata.

clear_session(session_id: str)

Delete all turns for a session and return the number removed.

start_new_session(address: ChatAddress)

Switch the active chat to a new session and return its id.

get_active_session_id(address: ChatAddress)

Return the current active session id for a chat address.

get_session_info(session_id: str)

Return command-facing session metadata.

get_session_run_status(session_id: str)

Return command-facing agent run status for a session.

list_commands()

List registered commands.

list_models()

List available provider/model pairs.

set_session_model(session_id: str, model_name: str)

Switch the session to a model and return provider id if found.

llm_chat(messages: list[dict[str, str]], model: str = '', temperature: float | None = None, max_tokens: int | None = None, tools: list[dict[str, Any]] | None = None)

Send a single-turn chat request to an LLM.

The model spec is resolved via the bot's built-in ModelRouter: it accepts tags ("cheap"), bare model names, or provider/model compound form. An empty string uses the default provider's default model.

Tools are passed through to the provider but NOT executed — the plugin receives tool_calls in the response and handles them itself.

run_subagent(prompt: str, model: str = '', system_prompt: str = '', tools: list[str] | None = None, max_steps: int = 10, timeout_seconds: int = 300)

Run a multi-turn subagent with optional tool access.

The subagent runs in an isolated child session. tools is a list of tool names to grant (empty = no tools). The subagent can call any of them during its reasoning loop.

Requires an active session context (e.g. inside a command or event handler).

update_runtime_settings(session_id: str, updates: dict[str, Any])

Merge runtime settings into session metadata and return the result.

memory_search(query: str, limit: int = 5)

Search the memory store for relevant records.

memory_store(key: str, content: str, metadata: dict[str, Any] | None = None)

Persist a record to the memory store.

search_chat_history(query: str, chat_address: str = '', role: str = '', limit: int = 20)

Search raw conversation turns across ALL sessions (soft-gated).

Returns dicts with session_id / role / content / created_at. Soft-gated (no permission check, no scope restriction) — memory is soft context; the gating lives in the calling tool's description.

search_chats(name: str, platform: str = '')

Fuzzy-search observed chat/group names → rows with chat_address.

Returns dicts with chat_address / display_name / platform / last_seen_at. Only knows chats the bot has seen (observe-only).

get_chat_names(chat_keys: list[str])

Bulk-resolve {chat_key: display_name} for observed chats.

Unseen keys are absent from the returned map. Empty map if unavailable.

plugin_data_get(key: str)

Read a value from this plugin's data store. Returns parsed JSON or None.

plugin_data_set(key: str, value: Any)

Write a value to this plugin's data store. Overwrites if the key exists.

plugin_data_delete(key: str)

Delete a key from this plugin's data store. Returns True if it existed.

plugin_data_list(prefix: str = '')

List key-value pairs, optionally filtered by key prefix.

workspace_read(path: str)

Read a file from the workspace. Subject to permission checks.

workspace_write(path: str, content: str)

Write a file to the workspace. Subject to permission checks.

resolve_workspace_path(path: str)

Resolve a workspace-relative path to an absolute local path.

get_workspace_root(workspace_id: str | None = None)

Return the filesystem root path for a workspace.

When workspace_id is None, uses the active workspace. Returns None when the workspace manager is unavailable.

publish_event(event: Any)

Publish an event on the event bus.

spawn_task(name: str, coro: Coroutine[Any, Any, Any], kind: str = 'oneshot')

Spawn a named background task owned by this plugin.

The task is automatically cancelled when the plugin is disabled. The coroutine runs in the bot's event loop. Uncaught exceptions are logged automatically.

cancel_task(name: str)

Cancel a previously spawned task by name. Returns True if found.

spawn_interval_task(name: str, func: Callable[[], Awaitable[None]], interval_seconds: float, initial_delay: float = 0.0)

Spawn a periodic task owned by this plugin.

get_document_store_manager()

Return the DocumentStoreManager for creating/accessing document collections.

Returns None if the document store subsystem is not available. Requires the llm_access permission.