agentscope.app.channelChannelBase
class ChannelBase(ABC)
agentscope/app/channel/_base.py:303–665平台渠道的抽象基类。位于 agentscope.app.channel 包,继承自 ABC。核心成员包括 start_listening、send_response、aclose、send_reaction、remove_reaction、list_bot_chats。
A subclass fully describes its platform type on the class itself, so
the service can register it from :func:~agentscope.app.create_app
(channels=[FeishuChannel, ...]) without a separate registry
table:
- :attr:
channel_type/ :attr:display_name/ :attr:platform_bot_id_field— type metadata; - nested :class:
Credentials/ :class:Config— the credential and option models, overridden by each subclass; the service renders their JSON Schema as frontend forms and validates against them; __init__(channel_id, credentials, config)— the uniform construction contract; instances are built per stored channel with that channel's validatedCredentials/Config.
嵌套类摘要Nested Class Summary
| 类Class | 类型Kind | 说明Description |
|---|---|---|
Credentials | 类class | 凭证,把智能体接入外部消息渠道与企业平台。 Secret connection fields (app id, tokens, ...). Subclasses override with their own fields; mark a field secret with ``json_schema_extra={"format": "password"}``. |
Config | 类class | 配置,保存相关配置项。 Non-secret per-channel behaviour switches. Subclasses override; left empty when the platform exposes no options. |
构造器摘要Constructor Summary
| 构造器Constructor | 说明Description |
|---|---|
def __init__(channel_id: str, credentials: 'ChannelBase.Credentials', config: 'ChannelBase.Config') -> None | 初始化 `ChannelBase` 实例。 Build a channel from its validated credentials and config. |
属性摘要Property Summary
| 属性Property | 说明Description |
|---|---|
channel_idabstractabstractmethod | 获取渠道标识。 The unique channel instance identifier. |
属性摘要Attribute Summary
| 属性Attribute | 类型Type | 说明Description | 值Value |
|---|---|---|---|
channel_type | str | 渠道类型字段。 | — |
display_name | str | 展示名称字段。 | — |
platform_bot_id_field | str | `platform_bot_id_field` 字段。 | — |
description | str | `description` 字段。 | '' |
icon_url | str | `icon_url` 字段。 | '' |
credential_binding | type['CredentialBindingBase'] | None | 凭证绑定字段。 | None |
capabilities | ChannelCapability | 能力字段。 | ChannelCapability() |
status | ChannelStatus | 状态字段。 | — |
方法摘要Method Summary
| 方法Method | 说明Description |
|---|---|
async def start_listening(emit: Callable[['ChannelEvent | ChannelConfirmationResultEvent'], Awaitable[None]]) -> Noneasyncabstractabstractmethod | 启动 `start_listening`。 Store ``emit``, set up resources, then connect and loop receiving events — normalising each into a ``ChannelEvent`` and ``await self._emit(event)`` (auto-reconnect) — releasing resources in a ``finally``. The sole connection-lifecycle method. |
async def send_response(event: ChannelEvent, events: AsyncIterator[dict]) -> Noneasyncabstractabstractmethod | 发送响应。 Consume the run's agent-event stream and deliver the reply. |
async def aclose() -> Noneasync | 返回 `aclose` 结果。 Release resources acquired outside :meth:`start_listening`. |
async def send_reaction(event: ChannelEvent, emoji_type: str) -> str | Noneasync | 发送 `send_reaction`。 Add an emoji reaction to the inbound message (e.g. "OnIt"). |
async def remove_reaction(event: ChannelEvent, reaction_id: str) -> Noneasync | 移除 `remove_reaction`。 Remove a reaction added by :meth:`send_reaction`. Default: no-op. |
async def list_bot_chats() -> list[dict]async | 列出 `list_bot_chats`。 Fetch the bot's chats/groups as dicts with at least ``chat_id`` and ``name``. Default: empty (unsupported). |
async def chat_kind(chat_id: str) -> 'ChatKind | None'async | 设置聊天种类。 Best-effort audience shape of ``chat_id`` (group vs 1:1), used to tailor the agent's session context. Default: ``None`` unknown. |
async def chat_name(chat_id: str) -> strasync | 设置聊天名称。 Best-effort human display name of ``chat_id`` (e.g. a group title), used to tailor the agent's context. Default: ``""``. |
async def list_tools(workspace: 'WorkspaceBase', channel_user_id: str | None = None) -> list['ToolBase']async | 列出当前连接可用的工具。 Platform tools exposed to the agent — e.g. send a file to a different user/group than the conversation. |
async def list_wiki_spaces(channel_user_id: str, limit: int, next_token: str | None = None) -> WikiPage[WikiSpace]async | 列出 `list_wiki_spaces`。 Wiki spaces readable by one platform user. |
async def list_wiki_nodes(channel_user_id: str, parent_node_id: str, limit: int, next_token: str | None = None) -> Wiki…async | 列出 `list_wiki_nodes`。 Direct children of one wiki space root or folder. |
async def read_wiki_document(channel_user_id: str, node_id: str, start_index: int, max_blocks: int) -> WikiDocument | N…async | 执行 `read_wiki_document` 方法。 Read a bounded range of one wiki document. |
构造器详情Constructor Detail
ChannelBase
def __init__(channel_id: str, credentials: 'ChannelBase.Credentials', config: 'ChannelBase.Config') -> None
初始化 ChannelBase 实例。
Build a channel from its validated credentials and config.
| 名称Name | 类型Type | 说明Description |
|---|---|---|
channel_id | str | 渠道标识(类型 `str`)。This channel instance's unique id. |
credentials | 'ChannelBase.Credentials' | 凭证(类型 `'ChannelBase.Credentials'`)。Validated secrets. |
config | 'ChannelBase.Config' | 配置对象(类型 `'ChannelBase.Config'`)。Validated platform options. |
agentscope/app/channel/_base.py:352–364属性详情Attribute Detail
channel_type
str渠道类型字段。声明为 str。
The channel type field.
agentscope/app/channel/_base.py:321display_name
str展示名称字段。声明为 str。
The display name field.
agentscope/app/channel/_base.py:324platform_bot_id_field
strplatform_bot_id_field 字段。声明为 str。
The platform bot id field field.
agentscope/app/channel/_base.py:327description
str = ''description 字段。声明为 str。
The description field.
agentscope/app/channel/_base.py:331icon_url
str = ''icon_url 字段。声明为 str。
The icon url field.
agentscope/app/channel/_base.py:334credential_binding
type['CredentialBindingBase'] | None = None凭证绑定字段。声明为 type['CredentialBindingBase'] | None。
The credential binding field.
agentscope/app/channel/_base.py:347capabilities
ChannelCapability = ChannelCapability()能力字段。声明为 ChannelCapability。
The capabilities field.
agentscope/app/channel/_base.py:366status
ChannelStatus状态字段。声明为 ChannelStatus。
The status field.
agentscope/app/channel/_base.py:368属性详情Property Detail
channel_id
def channel_id() -> str
获取渠道标识。
The unique channel instance identifier.
agentscope/app/channel/_base.py:388–389方法详情Method Detail
start_listening
async def start_listening(emit: Callable[['ChannelEvent | ChannelConfirmationResultEvent'], Awaitable[None]]) -> None
启动 start_listening。
Store emit, set up resources, then connect and loop receiving events — normalising each into a ChannelEvent and await self._emit(event) (auto-reconnect) — releasing resources in a finally.
| 名称Name | 类型Type | 说明Description |
|---|---|---|
emit | Callable[['ChannelEvent | ChannelConfirmationResultEvent'], Awaitable[None]] | 用于入站事件;存储 as`self._emit`和调用`await self._emit(event)`的网关回调。Gateway callback for inbound events; store as `self._emit` and call `await self._emit(event)`. |
agentscope/app/channel/_base.py:392–407send_response
async def send_response(event: ChannelEvent, events: AsyncIterator[dict]) -> None
发送响应。
Accumulate the events into a Msg (:meth:Msg.append_event, via :data:_EVENT_ADAPTER), render with :meth:_render, and send to the platform — streaming or one-shot as the channel chooses; present a confirmation when the run parks.
| 名称Name | 类型Type | 说明Description |
|---|---|---|
event | ChannelEvent | 发送目标(聊天标识)。The send target (chat id). |
events | AsyncIterator[dict] | 运行's 会话事件。The run's session events. |
agentscope/app/channel/_base.py:412–427aclose
async def aclose() -> None
返回 aclose 结果。
An instance built by :class:~agentscope.app.channel.ChannelClients never runs the connection loop, so whatever its REST calls opened lazily has no finally to close it.
agentscope/app/channel/_base.py:479–487send_reaction
async def send_reaction(event: ChannelEvent, emoji_type: str) -> str | None
发送 send_reaction。
Add an emoji reaction to the inbound message (e.g.
| 名称Name | 类型Type | 说明Description |
|---|---|---|
event | ChannelEvent | 入站事件到 ReAct 到。The inbound event to react to. |
emoji_type | str | emoji 类型(类型 `str`)。Platform emoji/reaction key. |
agentscope/app/channel/_base.py:489–504remove_reaction
async def remove_reaction(event: ChannelEvent, reaction_id: str) -> None
移除 remove_reaction。
Remove a reaction added by :meth:send_reaction.
| 名称Name | 类型Type | 说明Description |
|---|---|---|
event | ChannelEvent | 事件(类型 `ChannelEvent`)。The inbound event that was reacted to. |
reaction_id | str | reaction 标识(类型 `str`)。The id returned by :meth:`send_reaction`. |
agentscope/app/channel/_base.py:506–516list_bot_chats
async def list_bot_chats() -> list[dict]
列出 list_bot_chats。
Fetch the bot's chats/groups as dicts with at least chat_id and name.
agentscope/app/channel/_base.py:520–524chat_kind
async def chat_kind(chat_id: str) -> 'ChatKind | None'
设置聊天种类。
Best-effort audience shape of chat_id (group vs 1:1), used to tailor the agent's session context.
| 名称Name | 类型Type | 说明Description |
|---|---|---|
chat_id | str | 聊天标识(类型 `str`)。The platform chat to classify. |
agentscope/app/channel/_base.py:526–536chat_name
async def chat_name(chat_id: str) -> str
设置聊天名称。
Best-effort human display name of chat_id (e.g.
| 名称Name | 类型Type | 说明Description |
|---|---|---|
chat_id | str | 平台聊天到名称。The platform chat to name. |
agentscope/app/channel/_base.py:538–548list_tools
async def list_tools(workspace: 'WorkspaceBase', channel_user_id: str | None = None) -> list['ToolBase']
列出当前连接可用的工具。
Default: the wiki tools, for a platform that has a wiki and a session that acts as a known user.
| 名称Name | 类型Type | 说明Description |
|---|---|---|
workspace | 'WorkspaceBase' | 工作区(类型 `'WorkspaceBase'`)。The calling session's workspace, so file-sending tools read from it, not the host. |
channel_user_id默认default None | str | None | 渠道用户标识(类型 `str | None`)。The platform user the session acts as, for tools that read with that user's own permissions. `None` when the session has no single such user — a shared group session, or a chat whose audience the platform cannot classify. |
agentscope/app/channel/_base.py:552–581list_wiki_spaces
async def list_wiki_spaces(channel_user_id: str, limit: int, next_token: str | None = None) -> WikiPage[WikiSpace]
列出 list_wiki_spaces。
Every wiki call names the user it acts as, so the platform applies that user's own permissions rather than the bot's.
| 名称Name | 类型Type | 说明Description |
|---|---|---|
channel_user_id | str | 平台用户到读取 as。The platform user to read as. |
limit | int | 数量上限(类型 `int`)。Maximum spaces to return. |
next_token默认default None | str | None | 下一个令牌(类型 `str | None`)。Token from a previous page. |
agentscope/app/channel/_base.py:585–605list_wiki_nodes
async def list_wiki_nodes(channel_user_id: str, parent_node_id: str, limit: int, next_token: str | None = None) -> WikiPage[WikiNode]
列出 list_wiki_nodes。
Default: empty, for a platform with no wiki.
| 名称Name | 类型Type | 说明Description |
|---|---|---|
channel_user_id | str | 平台用户到读取 as。The platform user to read as. |
parent_node_id | str | 父级节点标识(类型 `str`)。The space root or folder to browse. |
limit | int | 数量上限(类型 `int`)。Maximum children to return. |
next_token默认default None | str | None | 下一个令牌(类型 `str | None`)。Token from a previous page. |
agentscope/app/channel/_base.py:607–627read_wiki_document
async def read_wiki_document(channel_user_id: str, node_id: str, start_index: int, max_blocks: int) -> WikiDocument | None
执行 read_wiki_document 方法。
Default: None, for a platform with no wiki.
| 名称Name | 类型Type | 说明Description |
|---|---|---|
channel_user_id | str | 平台用户到读取 as。The platform user to read as. |
node_id | str | 文档到读取。The document to read. |
start_index | int | 启动索引(类型 `int`)。First block index to read. |
max_blocks | int | 最大 blocks(类型 `int`)。Maximum blocks to read. |
- 可能抛出 `Exception`。`RuntimeError`: If the platform refuses the read, with a
- 可能抛出 `Exception`。message naming what the operator has to fix.
agentscope/app/channel/_base.py:629–654嵌套类详情Nested Class Detail
Credentials
class Credentials凭证,把智能体接入外部消息渠道与企业平台。位于 agentscope.app.channel 包。
Secret connection fields (app id, tokens, ...). Declared in agentscope.app.channel.
Config
class Config配置,保存相关配置项。位于 agentscope.app.channel 包。
Non-secret per-channel behaviour switches. Declared in agentscope.app.channel.