agentscope.app.channel.worker

A platform hands one bot's events to one connection: Feishu, DingTalk and Slack pick a single connection out of the ones a bot has open, while Discord broadcasts to all of them. So having every API replica connect either wastes connections or processes each message several times — the connections belong in a process whose count is chosen for them rather than for HTTP traffic.

This worker is that process. It owns:

• a :class:~agentscope.app.channel.ChannelLifecycleDispatcher, which reconciles the live connections against the stored channel records and heartbeats their status;
• a :class:~agentscope.app.channel.ChannelGateway, which turns each inbound event into a run trigger on the message bus.

It does not run agents, serve HTTP, or deliver replies — a reply is plain REST, so whichever node runs the agent sends it directly. Pair this with create_app(..., enable_channel_worker=False) so the API replicas hold no connections.

This module is a library: a deployment wires its concrete backends through :func:run_channel_worker from whatever bootstrap script it uses (systemd unit, Kubernetes Deployment, docker-compose command, ...). The shape mirrors :func:agentscope.app.create_app — pass already-constructed backend instances and the worker manages their lifecycle through an :class:AsyncExitStack.

Example::

import asyncio
import os

from agentscope.app.channel import FeishuChannel
from agentscope.app.channel.worker import run_channel_worker
from agentscope.app.message_bus import RedisMessageBus
from agentscope.app.storage import RedisStorage
from agentscope.app.workspace_manager import LocalWorkspaceManager

async def main() -> None:
    await run_channel_worker(
        storage=RedisStorage(url=os.environ["REDIS_URL"]),
        message_bus=RedisMessageBus(url=os.environ["REDIS_URL"]),
        workspace_manager=LocalWorkspaceManager(),
        channels=[FeishuChannel],
    )

if __name__ == "__main__":
    asyncio.run(main())