zulip-agent-adapter 0.1.5

Async Python adapter for connecting bots and AI agents to Zulip chat

Zulip Agent Adapter

A lightweight async Python library for connecting bots and AI agents to Zulip. It handles the event-loop plumbing — long-polling, reconnection, rate limits, deduplication — so you can focus on your bot's logic.

Status: Alpha (v0.1.5) — used in production with Hermes Agent, but the public API may still change based on feedback.

Features

• Async I/O via aiohttp — non-blocking event handling
• Auto-reconnection with exponential backoff and jitter
• Rate-limit handling (respects Zulip's retry-after on 429s)
• Sticky topic focus — bot stays engaged in a conversation after @-mention
• Configurable mention gating per channel
• Supports both channel (stream) and direct messages

Installation

From PyPI:

pip install zulip-agent-adapter

For development:

git clone https://github.com/carlh04426/zulip-agent-adapter.git
cd zulip-agent-adapter
pip install -e ".[dev]"

Quick Start

This example shows a simple echo bot that connects to Zulip using explicit config:

import asyncio
from zulip_agent_adapter import ZulipAdapter, ZulipConfig


async def main() -> None:
    config = ZulipConfig(
        base_url="https://your-org.zulipchat.com",
        bot_email="[e-mail]",
        api_key="your-api-key",
        chatmode="oncall",  # Require @mention to activate
    )

    adapter = ZulipAdapter(config)

    async def on_message(event):
        print(f"[{event.source.chat_name}] {event.source.user_name}: {event.text}")

        await adapter.send(
            event.source.chat_id,
            f"Echo: {event.text}",
            topic=event.source.chat_topic,
        )

    adapter.on_message = on_message

    connected = await adapter.connect()
    if not connected:
        raise SystemExit("Failed to connect to Zulip")

    print("Connected! Listening for messages... (Ctrl+C to stop)")
    try:
        while True:
            await asyncio.sleep(1)
    except KeyboardInterrupt:
        print("Disconnecting...")
        await adapter.disconnect()


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

You can also load configuration from environment variables using ZulipConfig.from_env() — see Configuration, immediately below for details.

Configuration

Required Settings

Setting	Description
base_url	Your Zulip server URL (e.g. https://acme.zulipchat.com)
bot_email	e.g. [e-mail]
api_key	Bot API key from Zulip settings

Behavior Settings

Setting	Default	Description
chatmode	"oncall"	"oncall" (require mention), "onmessage" (all messages), or "onchar" (accepted, behaves like onmessage — streaming not yet implemented)
require_mention	None	Override mention requirement (None = follow chatmode)
default_topic	"general"	Default topic for channel messages
streams	"*"	Comma-separated channel names to monitor, or "*" for all
free_response_streams	""	Comma-separated channel IDs that don't require mention
sticky_topic_after_mention	None	Override sticky focus (None = auto-enable for oncall)
sticky_topic_idle_minutes	45	Minutes before sticky focus expires

Environment Variables

ZulipConfig.from_env() reads configuration from the environment, which is useful for Docker deployments and CI:

export ZULIP_URL="https://your-org.zulipchat.com"
export ZULIP_BOT_EMAIL="[e-mail]"
export ZULIP_API_KEY="your-api-key"
export ZULIP_CHATMODE="oncall"       # optional, default: oncall
export ZULIP_STREAMS="general,dev"   # optional, default: *

from zulip_agent_adapter import ZulipConfig, ZulipAdapter

config = ZulipConfig.from_env()
adapter = ZulipAdapter(config)

Required variables (ZULIP_URL, ZULIP_BOT_EMAIL, ZULIP_API_KEY) must be set or from_env() will raise KeyError.

Concepts

Chat Modes

• oncall (default): Bot only responds when @-mentioned. Good for shared channels.
• onmessage: Bot responds to every message. Good for dedicated bot channels.
• onchar: Accepted but currently behaves like onmessage. Character-level streaming is planned.

Sticky Topic Focus

When enabled (on by default in oncall mode), after a user @-mentions the bot in a channel topic, the bot stays "focused" on that topic for a configurable period (default 45 minutes). During this window it responds to all messages in that topic without requiring additional @-mentions.

Users can clear focus with /unfocus or @**Bot Name** unfocus.

Message Targeting

Channel messages use stream:CHANNEL_ID:

await adapter.send("stream:123", "Hello!", topic="greetings")

Direct messages use private:USER_ID1,USER_ID2:

await adapter.send("private:456,789", "Hello!")

API Reference

ZulipAdapter

The main adapter class.

Methods

• async connect() -> bool — Connect to Zulip and start event polling
• async disconnect() — Disconnect and clean up resources
• async send(chat_id, content, topic=None) -> SendResult — Send a message
• async edit_message(message_id, content) -> SendResult — Edit an existing message
• async get_chat_info(chat_id) -> dict — Returns {"name": str, "type": "channel" | "dm"}

Callbacks

• on_message: Callable[[MessageEvent], Awaitable[None]] — Called for each incoming message

MessageEvent

Event object passed to on_message:

Attribute	Type	Description
text	str	Message text (bot mentions stripped in oncall mode)
message_type	MessageType	TEXT or COMMAND (starts with /)
source	MessageSource	Chat and user context (see below)
message_id	str	Zulip message ID
raw_message	dict	Full Zulip event payload for advanced use
reply_to_message_id	Optional[str]	ID of the triggering message

MessageSource

Attribute	Type	Description
chat_id	str	Routing ID ("stream:123" or "private:456")
chat_name	str	Human-readable channel or DM label
chat_type	str	"channel" or "dm"
user_id	Optional[str]	Sender's Zulip user ID
user_name	str	Sender's display name
thread_id	Optional[str]	Topic name (channels only)
chat_topic	Optional[str]	Same as thread_id; convenience alias

ZulipConfig

See Configuration above. Also provides ZulipConfig.from_env().

Testing

# Run all tests
pytest

# With coverage
pytest --cov=zulip_agent_adapter --cov-report=html

# Specific test file
pytest tests/test_adapter.py -v

Contributing

This adapter was originally developed as part of an internal Hermes-style agent gateway and then refactored into a standalone package.

If you’d like to contribute:

• Fork the repository
• Create a feature branch: git checkout -b feature/my-change
• Make your changes
• Run the checks: pytest, black src tests, ruff check src tests, mypy src
• Commit and push
• Open a Pull Request

If you're unsure about a change, feel free to open an issue to discuss it first.

License

MIT License — see LICENSE.

Acknowledgments

• Inspired by the OpenClaw gateway architecture.
• Originally developed for a Hermes-style agent gateway and extracted into this standalone adapter.