harness-gateway 0.8.3

Multi-platform IM channel bridge with unified message abstraction for AI agents and bots

多平台 IM 通道桥接 —— 用一套抽象接口，让 AI Agent 连接任意即时通讯平台。

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

	Feature	Description
🔌	9 platforms, one processor	Feishu, DingTalk, QQ, WeCom, WeChat iLink, Yuanbao, Xiaoyi, MQTT, Telegram — 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

📌 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

Layer	Technology
Language	Python 3.11+
Channel model	BaseChannel + per-platform adapter (9 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	✅	✅

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.

🛠️ 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

🔗 Related projects

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

📄 License

This project is licensed under the MIT License.