> ## Documentation Index
> Fetch the complete documentation index at: https://docs.agentscope.io/llms.txt
> Use this file to discover all available pages before exploring further.
# Custom channel
> Implement ChannelBase to connect a platform beyond the built-ins.
To connect a platform beyond the built-in Feishu and Discord, subclass `ChannelBase` to write a channel class and hand it to `create_app`. A channel class is the translation layer between the IM platform and the agent service: all platform differences are contained here, and the framework handles the orchestration.
A channel class does two things: **describe its type** (declare the type id, credentials, and config) and **implement its behavior** (keep the connection, normalize inbound messages and emit them, send replies back to the platform).
## Type description
A channel class carries its type information on the class itself, so the framework renders the frontend form, validates input, and constructs instances from it, with no separate registry.
```python A self-describing channel class theme={null}
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
```
The framework builds every instance with a uniform `(channel_id, credentials, config)`: `credentials` and `config` are already validated against the `Credentials` / `Config` you declared. `Credentials` holds secrets (encrypted, redacted, immutable); `Config` holds non-secret switches (plaintext, updatable); users fill in both per channel in the management UI.
## Required methods
Besides the constructor, `ChannelBase` has three abstract methods you must implement; the rest have sensible defaults (a no-op or "not supported").
| Method | Responsibility |
| ------------------------------ | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `channel_id` (property) | Return this channel instance's unique id |
| `start_listening(emit)` | Store `emit`, connect, and loop receiving; normalize each platform message into a `ChannelEvent` and `await self._emit(event)`. Include auto-reconnect, and release resources in a `finally` |
| `send_response(event, events)` | Consume one run's agent event stream `events`, accumulate a reply, and send it back to the platform |
## Receive messages
The framework calls `start_listening(emit)` when it starts the channel, passing in the inbound callback `emit`. The channel stores it as `self._emit`, normalizes each platform message into a `ChannelEvent`, and calls it; the framework handles the rest. A channel class neither holds nor imports the orchestration layer.
```python Store emit and normalize inbound messages theme={null}
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
```
`ChannelEvent.content` reuses the same `TextBlock` / `DataBlock` types as `Msg.content`, so multimodal messages reach the agent with no extra conversion.
## Send replies
The framework doesn't hand the channel a finished reply; it hands `send_response` the run's **event stream** and lets the channel accumulate, render, and send. That way a channel can send one complete reply, or stream updates as Feishu does. The method has two parameters: `event` is the send target, used only for its `chat_id` to locate which chat to reply to; `events` is the agent event stream this run produces. The base class provides `_render()`, which folds an accumulated reply into deliverable text / data blocks and, per the channel's display switches, decides whether to include the thinking process and tool calls.
```python Accumulate the event stream and send it back theme={null}
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
```
For a complete streaming and error-handling implementation, see the built-in `FeishuChannel` and `DiscordChannel`.
## Connection status
Each channel instance creates its own `self.status = ChannelStatus()` in `__init__` and updates `status.state` as it connects, reconnects, and stops (values `stopped` / `connecting` / `connected` / `retrying` / `failed`). This is exactly what the management UI and `GET /channels/{id}/status` read. If the first connection keeps failing, you can set `state` to `failed` and park, waiting for the user to change the config before reconnecting.
## Capability declaration
`capabilities` is the channel's declaration of what the platform supports, used by the framework and the channel itself when sending. Declare it truthfully for your platform:
```python Declare capabilities theme={null}
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
)
```
| Field | Meaning |
| -------------------- | --------------------------------------------------------------------------------------------- |
| `text` / `markdown` | Whether plain text / Markdown is supported |
| `image` / `file` | Whether images / files are supported; unsupported outbound media degrades to placeholder text |
| `interactive` | Whether the platform can present interactive confirmation UI |
| `streaming` | Whether a reply can be updated within a single message |
| `max_message_length` | Per-message character cap; `_split_long_message()` splits by it automatically |
## Tool confirmation
When an agent calls a tool that needs approval, the run pauses and a `RequireUserConfirmEvent` appears in the event stream. The channel recognizes it in `send_response` and presents the request to the user: capable platforms use an interactive card or buttons, and a plain-text platform can send a "reply yes/no" prompt. `RequireUserConfirmEvent.tool_calls` is the list of tools awaiting approval, each with `id`, `name`, and `input`.
Once the user decides, the channel normalizes it into a `ChannelConfirmationResultEvent` and emits it through the same `self._emit` entry as regular messages:
```python Present the confirmation and return the decision theme={null}
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,
),
)
```
`ChannelConfirmationResultEvent` carries only lookup keys: the authoritative awaiting tool call is read from session state on resume, never trusted from the card. This makes the round-trip distributed-safe by nature, no matter how long the user waits to click or which node the click lands on.
When you present the confirmation, the `event.metadata` that `send_response` received carries the run's `agent_id` and `session_id`. Include them in the `ChannelConfirmationResultEvent` so a click resumes exactly the session that asked for approval, with no need to re-match routing rules.
## Optional methods
These all have default implementations; override them per your platform's capabilities:
| Method | Purpose |
| --------------------------------------- | --------------------------------------------------------------------------------------------- |
| `send_reaction()` / `remove_reaction()` | Add / remove an emoji reaction on an inbound message (e.g. "working") |
| `list_bot_chats()` | Return the chats the bot is in, for the management UI to pick from when configuring routing |
| `chat_kind()` / `chat_name()` | Return whether a chat is a group or a DM, and its name, to enrich the agent's session context |
| `list_tools()` | Expose platform-specific tools to the agent (e.g. send a file to a specific user / group) |
## Enable the channel
Add your channel class to `create_app`'s `channels`, and the service accepts that platform. `channels` declares **which channel types the whole service accepts**; omitting it enables none, so list the built-in types and your custom types together.
```python Enable in create_app theme={null}
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],
)
```
Once enabled, `MyChannel` shows up in the platform-type list of the management UI, and users can fill in credentials, configure routing, and create channels just like a built-in platform.
## Further reading
Custom channels reuse the same routing model.
Back to how channels work overall.