harness-gateway 0.9.8

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

Highlights · Overview · Core Technology · Features · Quick Start · Contents

English · 中文

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

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

✨ Highlights

📌 Overview

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.

🧠 Core Technology

🤔 Features

Supported platforms

Unified message abstraction

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

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.

🚀 Quick Start

Prerequisites

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

1. Install

# Core library
pip install harness-gateway

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

2. Minimal echo bot (Telegram)

import asyncio, os
from collections.abc import AsyncIterator

from harness_gateway import ChannelManager, InboundMessage, MessageEvent
from harness_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 harness_gateway import ChannelManager, InboundMessage, MessageEvent
from harness_gateway.channels.dingtalk import DingTalkConfig
from harness_gateway.channels.feishu import FeishuConfig
from harness_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.

📑 Contents

• Highlights
• Overview
• Core Technology
• Features
• Quick Start
• 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 harness_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.11+, 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).

🔗 Related projects

📄 License

This project is licensed under the MIT License.