自定义渠道

from pydantic import BaseModel, Field
from agentscope.app.channel import ChannelBase


class MyChannel(ChannelBase):
    channel_type = "my_platform"       # 唯一类型标识
    display_name = "My Platform"       # 管理界面显示名
    platform_bot_id_field = "token"    # 用哪个凭据字段做接入查重
    description = "把智能体接入 My Platform"    # 管理界面一句话描述（可选）
    icon_url = "https://example.com/icon.png"  # 管理界面品牌图标（可选）

    class Credentials(BaseModel):      # 密钥字段，前端据此渲染凭据表单
        token: str = Field(
            title="Bot Token",
            json_schema_extra={"format": "password"},  # 标记为机密
        )

    class Config(BaseModel):           # 非密钥开关，没有可留空
        only_at_reply: bool = True

    def __init__(
        self,
        channel_id: str,
        credentials: "MyChannel.Credentials",
        config: "MyChannel.Config",
    ) -> None:
        self._channel_id = channel_id
        self._token = credentials.token          # 从校验后的凭据读取
        self._config = config                    # 校验后的非密钥配置

from collections.abc import Awaitable, Callable

from agentscope.app.channel import (
    ChannelConfirmationResultEvent,
    ChannelEvent,
    ChannelStatus,
)
from agentscope.message import TextBlock

# 入站回调：把规范化后的事件交给框架处理
Emit = Callable[[ChannelEvent | ChannelConfirmationResultEvent], Awaitable[None]]


async def start_listening(self, emit: Emit) -> None:
    self._emit = emit
    self.status = ChannelStatus(state="connecting")
    try:
        async for raw in self._connect():            # 平台长连接
            self.status.state = "connected"
            await self._emit(
                ChannelEvent(
                    channel_id=self._channel_id,
                    channel_user_id=raw.user_id,         # 平台侧用户标识
                    chat_id=raw.chat_id,                 # 驱动会话归组与路由匹配
                    content=[TextBlock(text=raw.text)],   # 与 Msg.content 同构
                    metadata={"chat_type": raw.chat_type},  # 供路由 match_key 匹配
                ),
            )
    finally:
        self.status.state = "stopped"                # 释放资源

from collections.abc import AsyncIterator

from pydantic import TypeAdapter

from agentscope.app.channel import ChannelEvent
from agentscope.event import AgentEvent, RequireUserConfirmEvent
from agentscope.message import Msg

_ADAPTER = TypeAdapter(AgentEvent)


async def send_response(
    self,
    event: ChannelEvent,             # 发送目标：用它的 chat_id 定位聊天
    events: AsyncIterator[dict],      # 这次运行的智能体事件流，逐个到达
) -> None:
    reply: Msg | None = None
    async for raw in events:
        evt = _ADAPTER.validate_python(raw)          # 还原为事件对象
        if isinstance(evt, RequireUserConfirmEvent):
            await self._present_confirm(event, evt)   # 需要审批，见下节
            return
        if reply is None:
            reply = Msg(name="assistant", role="assistant", content=[])
            reply.id = evt.reply_id
        reply.append_event(evt)                      # 累积成一条回复
    blocks = self._render(reply)                     # 折叠成可发送的块
    await self._deliver(event.chat_id, blocks)       # 你的平台发送实现

from agentscope.app.channel import ChannelCapability


class MyChannel(ChannelBase):
    capabilities = ChannelCapability(
        text=True,
        markdown=True,
        image=True,
        file=True,
        interactive=True,          # 平台能否呈现交互式确认 UI
        streaming=True,            # 能否在一条消息内流式更新
        max_message_length=4000,   # 单条消息字符上限
    )

from agentscope.app.channel import ChannelConfirmationResultEvent, ChannelEvent
from agentscope.event import RequireUserConfirmEvent


async def _present_confirm(
    self,
    event: ChannelEvent,
    req: RequireUserConfirmEvent,
) -> None:
    for tool in req.tool_calls:                   # 每个待确认的工具
        await self._send_buttons(
            event.chat_id,
            text=f"是否允许执行工具 {tool.name}？",
            value=tool.id,                        # 把 tool_call_id 塞进按钮，点击原样带回
        )


async def _on_button_click(self, click) -> None:
    await self._emit(
        ChannelConfirmationResultEvent(
            channel_id=self._channel_id,
            chat_id=click.chat_id,
            channel_user_id=click.user_id,        # 点击者
            tool_call_id=click.value,             # 原样回传，渠道不需理解其含义
            approved=click.approved,
        ),
    )

from agentscope.app import create_app
from agentscope.app.channel import DiscordChannel, FeishuChannel

app = create_app(
    storage=...,
    message_bus=...,
    workspace_manager=...,
    # 允许接入的渠道类型：内置两个 + 你的自定义类型
    channels=[FeishuChannel, DiscordChannel, MyChannel],
)