Custom channel

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


class MyChannel(ChannelBase):
    channel_type = "my_platform"       # unique type id
    display_name = "My Platform"       # name shown in the management UI
    platform_bot_id_field = "token"    # credential field used to de-duplicate connections
    description = "Connect agents to My Platform"  # one-line UI description (optional)
    icon_url = "https://example.com/icon.png"      # brand icon for the UI (optional)

    class Credentials(BaseModel):      # secret fields; the frontend renders the credential form
        token: str = Field(
            title="Bot Token",
            json_schema_extra={"format": "password"},  # mark as secret
        )

    class Config(BaseModel):           # non-secret switches; leave empty if none
        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          # read from the validated credentials
        self._config = config                    # the validated non-secret config

from collections.abc import Awaitable, Callable

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

# Inbound callback: hand a normalized event to the framework
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():            # platform long connection
            self.status.state = "connected"
            await self._emit(
                ChannelEvent(
                    channel_id=self._channel_id,
                    channel_user_id=raw.user_id,         # platform-side user id
                    chat_id=raw.chat_id,                 # drives session grouping and routing
                    content=[TextBlock(text=raw.text)],   # same shape as Msg.content
                    metadata={"chat_type": raw.chat_type},  # available to routing match_key
                ),
            )
    finally:
        self.status.state = "stopped"                # release resources

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,             # send target: use its chat_id to locate the chat
    events: AsyncIterator[dict],      # the agent event stream for this run, arriving one by one
) -> None:
    reply: Msg | None = None
    async for raw in events:
        evt = _ADAPTER.validate_python(raw)          # restore the event object
        if isinstance(evt, RequireUserConfirmEvent):
            await self._present_confirm(event, evt)   # needs approval, see below
            return
        if reply is None:
            reply = Msg(name="assistant", role="assistant", content=[])
            reply.id = evt.reply_id
        reply.append_event(evt)                      # accumulate into one reply
    blocks = self._render(reply)                     # fold into deliverable blocks
    await self._deliver(event.chat_id, blocks)       # your platform-send implementation

from agentscope.app.channel import ChannelCapability


class MyChannel(ChannelBase):
    capabilities = ChannelCapability(
        text=True,
        markdown=True,
        image=True,
        file=True,
        interactive=True,          # can the platform present interactive confirmation UI
        streaming=True,            # can it update a single message incrementally
        max_message_length=4000,   # per-message character cap
    )

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:                   # each tool awaiting approval
        await self._send_buttons(
            event.chat_id,
            text=f"Allow running the tool {tool.name}?",
            value=tool.id,                        # embed tool_call_id in the button, returned verbatim on click
        )


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,        # who clicked
            tool_call_id=click.value,             # returned verbatim; the channel need not understand it
            approved=click.approved,
        ),
    )

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

app = create_app(
    storage=...,
    message_bus=...,
    workspace_manager=...,
    # accepted channel types: the two built-ins + your custom type
    channels=[FeishuChannel, DiscordChannel, MyChannel],
)