> ## 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.
> ## Agent Instructions
> For AgentScope Python, use https://docs.agentscope.io/stable/en/index for new projects. For existing projects, check the installed agentscope version and use matching versioned documentation.
> The /latest/ alias points to development documentation. Use it only with the matching development source. Do not mix AgentScope 1.x and 2.x APIs.
> State the AgentScope version when providing installation commands or code examples. ReMe uses its own continuously updated /reme/latest/ documentation.
# 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
wiki=True, # whether the platform has a knowledge base the agent can read
)
```
| 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 |
| `wiki` | Whether the platform has a knowledge base the agent can browse and read. Unlike the flags above, this one describes the read direction |
A channel that declares `wiki=True` implements three methods on `ChannelBase`: `list_wiki_spaces`, `list_wiki_nodes` and `read_wiki_document`. Once they are there, the channel inherits the `ListWikiSpaces` / `ListWikiNodes` / `ReadWikiDocument` read-only tools that call them — nothing to register. Each method receives the platform user id of whoever sent the current message; make the request as that user and let the platform do the authorization. A run with no trusted sender receives none of these tools. See the [DingTalk knowledge base tools](/versions/2.0.9dev/en/deploy/channel/dingtalk#knowledge-base-tools) for a worked implementation.
## 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) |
## Authorize by QR Code
When a platform lets the operator hand over credentials elsewhere (scanning a QR code, clicking through a consent page), there is no need to make users copy an App Secret by hand. Attach a `credential_binding` to the channel class and the credential form in the management UI gains an "authorize by QR code" tab:
```python Declare a credential binding theme={null}
from typing import Any
from agentscope.app.channel import (
BindingState,
BindingStep,
ChannelBase,
CredentialBindingBase,
)
class MyCredentialBinding(CredentialBindingBase):
async def begin(self) -> BindingStep:
"""Open an authorization session and return the URL for the operator."""
payload = await request_device_code()
return BindingStep(
verification_url=payload["verification_uri"], # rendered as a QR code
provider_state={"device_code": payload["device_code"]}, # handed back verbatim
retry_after_secs=5, # polling interval the platform asks for
expires_in_secs=600, # lifetime of the session
)
async def advance(self, provider_state: dict[str, Any]) -> BindingStep:
"""Called once per poll: ask the platform whether the operator confirmed."""
payload = await poll_device_code(provider_state["device_code"])
if payload.get("client_secret"):
return BindingStep(
state=BindingState.AUTHORIZED,
credentials={"token": payload["client_secret"]}, # must match Credentials
)
# Still waiting: return the provider_state as is and the session continues
return BindingStep(provider_state=provider_state)
class MyChannel(ChannelBase):
channel_type = "my_platform"
credential_binding = MyCredentialBinding # without it, only manual entry is offered
```
Four constraints to respect when implementing one:
| Constraint | Description |
| ----------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| Stateless | A provider is instantiated once per process, and the two steps of one authorization may well land on different replicas. Anything needed between the steps (device code, PKCE verifier, the domain you were redirected to) goes into the returned `provider_state`, which the next step receives verbatim |
| Poll-driven | `advance()` is called once per client status query rather than looping on its own, so never wait or retry inside it |
| Credential fields | The `credentials` returned with `AUTHORIZED` must line up with the channel class's `Credentials` fields; the service creates the channel straight from them |
| Terminal states | `AUTHORIZED` / `FAILED` / `CANCELLED` are terminal and `advance()` is not called again; on failure put the reason in `error` so the operator sees it in the UI |
## Enable 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.