octop-gateway

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.

What is Octop Gateway?

	Feature	Description
🔌	10 platforms, one processor	Feishu, DingTalk, QQ, WeCom, WeChat iLink, Yuanbao, Xiaoyi, MQTT, Telegram, Discord — all behind one interface
🧩	Unified abstraction	BaseChannel turns each platform’s quirks into a common InboundMessage / MessageEvent model
📨	Streaming events	Your processor is an async generator yielding MessageEvent — first-class token streaming
💾	Pluggable media	MediaBackend stores attachments; FileSystemMediaBackend ships by default
🚦	Constraints	Per-channel rate limit, timeout, and typing indicator
📡	Push routing	push_text / push_content / push_to_all for proactive messages
🏢	Multi-tenant	Run multiple channels of the same kind, each isolated
🐍	Pythonic	Pure asyncio, fully typed models, no hidden magic

Why Octop Gateway?

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 MessageEvent s. 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.

Core Technology

Layer	Technology
Language	Python 3.12+
Channel model	BaseChannel + per-platform adapter (10 built-in)
Messaging model	InboundMessage / MessageEvent / ContentPart
Orchestration	ChannelManager (async queues, worker pools)
Media	MediaBackend (FileSystemMediaBackend default)
Constraints	Rate limit / timeout / typing indicator
Build / quality	hatchling · ruff · mypy · pytest

Features

Supported platforms

Platform	Channel kind	Transport	Text	Media
Feishu (Lark)	feishu	WebSocket + REST	✅	✅
DingTalk	dingtalk	Stream	✅	✅
QQ	qq	WebSocket	✅	✅
WeCom (Enterprise WeChat)	wecom	Callback + API	✅	✅
WeChat iLink	weixin	—	✅	✅
Yuanbao (元宝)	yuanbao	—	✅	✅
Xiaoyi (小艺)	xiaoyi	—	✅	✅
MQTT	mqtt	MQTT	✅	✅
Telegram	telegram	Long-polling	✅	✅

Discord

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.

Unified message abstraction

• InboundMessage carries the text, structured ContentPart s (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_*.

Custom channels

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

Constraints, media & push

• 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.

Multi-tenant

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

How to Use

Prerequisites

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

1. Install

# Core library
pip install octop-gateway

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

2. Minimal echo bot (Telegram)

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())

3. Add more platforms

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.

Documentation

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

Architecture

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 QR binding

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 conversation policy

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.

Development

Prerequisites: Python 3.12+, uv

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

🤝 Contributing

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).

Project	Description
octop-harness	Agent runtime that drives the gateway processor
octop-memory	Memory system for gateway-backed agents
octop-browser	Browser automation for agents
Octop	The self-hosted assistant that composes the Harness stack

📄 License

This project is licensed under the MIT License.

✨ Contributors

Thanks to all contributors: