Octop Gateway

Multi-platform IM channel bridge — one abstraction layer that lets AI agents connect to any instant-messaging platform.

What is Octop Gateway? · Why Octop Gateway? · How to Use · Documentation

English · 中文

octop-gateway is a multi-platform IM channel bridge with a unified message abstraction for AI agents and bots. It connects to mainstream IM platforms and normalizes every inbound message into one processing pipeline, so your agent logic is written once and runs across Feishu, DingTalk, QQ, WeCom, WeChat iLink, Yuanbao, Xiaoyi, MQTT, and Telegram.

octop-gateway's design goal: you write a single async message processor, and the gateway handles transport, parsing, media, and delivery for every platform behind a common interface.

Harness Gateway sits between your agent and the outside world. Each platform is implemented as a BaseChannel subclass that knows how to connect, parse inbound traffic, and send replies. A ChannelManager orchestrates them through async queues and hands your agent a normalized stream of MessageEvents. You never write platform-specific code in your bot — just one processor.

Because the abstraction lives in the gateway, swapping IM platforms is a configuration change, not a rewrite.

Use a Bot Token and enable Message Content Intent in the developer portal. allow_all_channels defaults to True, including configurations without this field, allowing all guild channels accessible to the bot. Set it to False to restrict access to allowed_channel_ids (empty denies guild messages). DMs always require allowed_user_ids; an empty user list denies DMs. Guild messages require a direct bot mention by default, and threads inherit parent-channel access. Supports HTTP proxying, attachments, typing and split replies. See examples/discord_bot.py.

• InboundMessage carries the text, structured ContentParts (text / image / video / audio / file), and a ChannelSubject.
• Your processor is a Callable[[InboundMessage], AsyncIterator[MessageEvent]] — emit MESSAGE for complete text, DELTA for token streaming, and COMPLETED to flush.
• ChannelConfig is a typed dataclass per platform; BaseChannel defines start / stop / parse_inbound / _send_*.

Subclass BaseChannel, register it with ChannelManager.add_channel(...), and the rest of the pipeline (media, constraints, push) works unchanged.

• Constraints — rate limit, response timeout, and typing indicator per channel.
• Media — pluggable MediaBackend; persist attachments wherever you like.
• Push — push_text / push_content to one subject, or push_to_all for broadcasts.

Run several channels of the same kind (e.g. two Feishu apps for two teams) — each is isolated by channel_id.

• Python 3.12+
• Credentials for the platforms you connect to

# Core library
pip install octop-gateway

# With example / agent integration extras
pip install "octop-gateway[examples]"

import asyncio, os
from collections.abc import AsyncIterator

from octop_gateway import ChannelManager, InboundMessage, MessageEvent
from octop_gateway.channels.telegram import TelegramConfig


async def echo(message: InboundMessage) -> AsyncIterator[MessageEvent]:
    yield MessageEvent.text(f"Echo: {message.text}")
    yield MessageEvent.completed()


async def main():
    manager = ChannelManager(processor=echo)
    await manager.start()
    await manager.add_telegram_channel(TelegramConfig(bot_token=os.environ["TELEGRAM_BOT_TOKEN"]))
    await asyncio.Event().wait()


asyncio.run(main())

import asyncio, os
from collections.abc import AsyncIterator

from octop_gateway import ChannelManager, InboundMessage, MessageEvent
from octop_gateway.channels.dingtalk import DingTalkConfig
from octop_gateway.channels.feishu import FeishuConfig
from octop_gateway.channels.qq import QQConfig


async def unified_bot(msg: InboundMessage) -> AsyncIterator[MessageEvent]:
    yield MessageEvent.text(f"[{msg.channel_type}] {msg.text}")
    yield MessageEvent.completed()


async def main():
    manager = ChannelManager(processor=unified_bot, workers_per_channel=4)
    await manager.start()

    await manager.add_feishu_channel(
        FeishuConfig(app_id=os.environ["FEISHU_APP_ID"], app_secret=os.environ["FEISHU_APP_SECRET"])
    )
    await manager.add_qq_channel(
        QQConfig(app_id=os.environ["QQ_APP_ID"], token=os.environ["QQ_TOKEN"], secret=os.environ["QQ_SECRET"])
    )
    await manager.add_dingtalk_channel(
        DingTalkConfig(app_key=os.environ["DINGTALK_APP_KEY"], app_secret=os.environ["DINGTALK_APP_SECRET"])
    )
    await asyncio.Event().wait()


asyncio.run(main())

Copy .env.example to .env for environment-based configuration.

• What is Octop Gateway?
• Why Octop Gateway?
• How to Use
• Reference
  • Architecture
  • Development
• Project Info
  • Contributing
  • Related projects
  • License

ChannelManager
 ├─ async queues + worker pools (per channel)
 ├─ add_channel(BaseChannel) / add_*_channel(...)
 ├─ push_text / push_content / push_to_all
 └─ per-channel BaseChannel
      ├─ start / stop
      ├─ parse_inbound → InboundMessage
      └─ _send_text / _send_content / _send_media

MessageProcessor: InboundMessage → AsyncIterator[MessageEvent]

Each BaseChannel owns its transport; the manager owns scheduling, media, constraints, and fan-out. Your processor only sees the normalized stream.

QQ Bot credentials can be obtained without manually copying an AppID and AppSecret. Render the returned URL as a QR code, then wait for confirmation:

from octop_gateway.channels.qq import QQBotQRLogin, QQConfig

login = QQBotQRLogin(source="octop")
qr = await login.fetch_qr_code()
print(qr.qrcode_url)  # Render this URL as a QR code in your UI or terminal.

result = await login.wait_for_login(qr.task_id)
if not result.connected:
    raise RuntimeError(result.message)
config = QQConfig.from_qr_credentials(result.credentials[0])

The one-time decryption key remains in memory and is discarded after success, expiry, cancellation, or timeout. Persisting the returned channel config is the caller's responsibility.

group_context separates platform visibility from agent activation. QQ enables the shared manager by default with a conservative auto + mention + recent(10) policy. It can be configured globally and overridden by native group ID:

{
  "group_context": {
    "enabled": true,
    "visibility": "auto",
    "activation": "mention",
    "history": "recent",
    "history_limit": 10,
    "history_ttl_seconds": 300,
    "clear_after_reply": true,
    "groups": {
      "GROUP_OPENID_WITH_FULL_ACCESS": {
        "visibility": "all",
        "activation": "always"
      },
      "GROUP_OPENID_MENTION_ONLY": {
        "visibility": "mention_only",
        "history": "none"
      }
    }
  }
}

• visibility: auto, all, mention_recent, or mention_only.
• activation: mention or always. always is accepted only with explicit all visibility, so replayed recent messages cannot trigger multiple replies.
• history: recent or none. Passive messages are bounded, expire by TTL, and are cleared after a successful reply.

When a platform permission is downgraded, update visibility to the granted level (or restart/reconfigure the channel); this immediately drops the old in-memory buffer. Platforms that do not publish permission-change events cannot be detected perfectly, so auto remains mention-triggered and the TTL limits stale context.

Prerequisites: Python 3.12+, uv

make install          # pip install -e ".[dev,examples]"
make all              # lint + typecheck + test

1. Fork the repository
2. Create a feature branch (git checkout -b feature/amazing-feature)
3. Run make all before submitting
4. Open a Pull Request against main

Channel conventions: AGENTS.md.

See CONTRIBUTING.md for branching, PR, and release details (release/* → main auto-publishes to PyPI).

This project is licensed under the MIT License.

Thanks to all contributors: