Build Your Own Agent

This guide walks you through building a custom AI agent that listens for messages in Zenzap and responds using OpenAI. By the end, you’ll have a working bot that polls for new messages, processes them through an LLM, and sends replies back to your topics.

Prerequisites

A Zenzap API key and secret (see Authentication)
Python 3.10+
An OpenAI API key

Project Setup

Create a new directory for your agent and install the required dependencies:

mkdir zenzap-agent && cd zenzap-agent
python -m venv venv && source venv/bin/activate
pip install requests python-dotenv openai

Create a .env file with your credentials:

BOT_API_KEY=your_bot_api_key_here
BOT_SECRET=your_bot_secret_here
CONTROL_CHANNEL_ID=your_control_channel_topic_id_here
OPENAI_API_KEY=your_openai_api_key_here
OPENAI_MODEL=gpt-4o
API_BASE_URL=<from_console>  # provided when you create the agent

Variable	Description
`BOT_API_KEY`	Your Zenzap API key (from the console)
`BOT_SECRET`	Your Zenzap API secret (used for request signing)
`CONTROL_CHANNEL_ID`	The topic ID where the bot will post status messages (e.g. “connected”, “disconnecting”)
`OPENAI_API_KEY`	Your OpenAI API key
`OPENAI_MODEL`	The OpenAI model to use (defaults to `gpt-4o`)
`API_BASE_URL`	The Zenzap API base URL — provided by the Zenzap console when you create the agent. It is region-specific; do not hardcode `https://api.zenzap.co`.

Step 1 — Build the Zenzap API Client

Create zenzap_client.py. This module handles all communication with the Zenzap API, including HMAC request signing (see Authentication for details). Start with the response wrapper and client constructor:

import hashlib
import hmac
import json
import time
from typing import Any, Optional
from dataclasses import dataclass
from urllib.parse import quote, urlencode

import requests


@dataclass
class ApiResponse:
    status: int
    data: Any
    success: bool

    @classmethod
    def from_response(cls, response: requests.Response) -> "ApiResponse":
        try:
            data = response.json()
        except ValueError:
            text = response.text.strip()
            data = {"raw": text} if text else {}
        return cls(
            status=response.status_code,
            data=data,
            success=200 <= response.status_code < 300
        )

    @classmethod
    def from_exception(cls, exception: requests.RequestException) -> "ApiResponse":
        return cls(
            status=0,
            data={"error": str(exception), "type": exception.__class__.__name__},
            success=False,
        )


class ZenzapClient:
    def __init__(
        self,
        api_key: str,
        secret: str,
        base_url: str = "https://api.zenzap.co",
        timeout: float = 30.0,
    ):
        self.api_key = api_key
        self.secret = secret
        self.base_url = base_url.rstrip("/")
        self.timeout = timeout

Next, add the private methods that handle request signing and HTTP calls. Every request must include an HMAC-SHA256 signature — GET requests sign the URI path, while POST/PATCH/DELETE requests sign the JSON body:

    def _generate_signature(self, data: str, timestamp: str) -> str:
        return hmac.new(
            self.secret.encode("utf-8"),
            f"{timestamp}.{data}".encode("utf-8"),
            hashlib.sha256
        ).hexdigest()

    def _get_headers(self, signature: str, timestamp: str, include_content_type: bool = False) -> dict:
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "X-Signature": signature,
            "X-Timestamp": timestamp,
        }
        if include_content_type:
            headers["Content-Type"] = "application/json"
        return headers

    def _get(self, path: str) -> ApiResponse:
        timestamp = str(int(time.time() * 1000))
        signature = self._generate_signature(path, timestamp)
        url = f"{self.base_url}{path}"
        try:
            response = requests.get(
                url,
                headers=self._get_headers(signature, timestamp),
                timeout=self.timeout,
            )
            return ApiResponse.from_response(response)
        except requests.RequestException as exception:
            return ApiResponse.from_exception(exception)

    def _request_with_body(self, method: str, path: str, body: dict) -> ApiResponse:
        body_str = json.dumps(body, separators=(",", ":"))
        timestamp = str(int(time.time() * 1000))
        signature = self._generate_signature(body_str, timestamp)
        url = f"{self.base_url}{path}"
        try:
            response = requests.request(
                method,
                url,
                headers=self._get_headers(signature, timestamp, include_content_type=True),
                data=body_str,
                timeout=self.timeout,
            )
            return ApiResponse.from_response(response)
        except requests.RequestException as exception:
            return ApiResponse.from_exception(exception)

    def _post(self, path: str, body: dict) -> ApiResponse:
        return self._request_with_body("POST", path, body)

    def _patch(self, path: str, body: dict) -> ApiResponse:
        return self._request_with_body("PATCH", path, body)

    def _delete(self, path: str, body: dict) -> ApiResponse:
        return self._request_with_body("DELETE", path, body)

Finally, add the public API methods your agent will use:

    def get_current_member(self) -> ApiResponse:
        return self._get("/v2/members/me")

    def get_topic(self, topic_id: str) -> ApiResponse:
        return self._get(f"/v2/topics/{topic_id}")

    def send_message(self, topic_id: str, text: str, external_id: Optional[str] = None) -> ApiResponse:
        body = {"topicId": topic_id, "text": text}
        if external_id:
            body["externalId"] = external_id
        return self._post("/v2/messages", body)

    def mark_message_read(self, message_id: str) -> ApiResponse:
        return self._post(f"/v2/messages/{message_id}/read", {})

    def add_reaction(self, message_id: str, reaction: str) -> ApiResponse:
        return self._post(f"/v2/messages/{message_id}/reactions", {"reaction": reaction})

    def get_updates(
        self,
        offset: Optional[str] = None,
        limit: int = 100,
        poll_timeout: int = 30,
    ) -> ApiResponse:
        params: dict[str, Any] = {"limit": limit, "timeout": poll_timeout}
        if offset:
            params["offset"] = offset
        query = urlencode({k: v for k, v in params.items() if v is not None}, doseq=True)
        path = f"/v2/updates?{query}" if query else "/v2/updates"
        timestamp = str(int(time.time() * 1000))
        signature = self._generate_signature(path, timestamp)
        url = f"{self.base_url}{path}"
        try:
            response = requests.get(
                url,
                headers=self._get_headers(signature, timestamp),
                timeout=poll_timeout + 10,
            )
            return ApiResponse.from_response(response)
        except requests.RequestException as exception:
            return ApiResponse.from_exception(exception)

The client uses additional methods like create_topic, list_topics, create_task, etc. See the full API Reference endpoints for everything you can do.

Step 2 — Handle Incoming Messages

Create bot.py. Start by defining a system prompt and a state object to track the bot’s runtime data:

import os
import signal
import sys
import time
from dataclasses import dataclass, field
from typing import Optional

from dotenv import load_dotenv
from openai import OpenAI

from zenzap_client import ZenzapClient

DEFAULT_SYSTEM_PROMPT = (
    "You are an AI assistant embedded in Zenzap, a team messaging platform. "
    "You operate across multiple topics (group chats). Each message you receive "
    "includes the sender's name and topic name as context. "
    "Be concise — this is chat, not a document. Short answers win. Expand only when asked. "
    "Never start with filler like 'Great question!' or 'Sure!'. Just answer. "
    "You have no memory between topics unless explicitly told."
)


@dataclass
class BotState:
    bot_member_id: str
    next_offset: Optional[str] = None
    topic_name_cache: dict = field(default_factory=dict)
    conversation_histories: dict = field(default_factory=dict)
    running: bool = True

bot_member_id — the bot’s own user ID, so it can skip its own messages.
next_offset — the cursor for long polling.
topic_name_cache — avoids repeated API calls to resolve topic names.
conversation_histories — per-topic message history sent to OpenAI for context.

Now add the core message handler. When a message arrives, it appends it to the topic’s conversation history, calls OpenAI, and sends the reply back:

def resolve_topic_name(state: BotState, zenzap: ZenzapClient, topic_id: str) -> str:
    if topic_id not in state.topic_name_cache:
        resp = zenzap.get_topic(topic_id)
        state.topic_name_cache[topic_id] = resp.data.get("name", topic_id) if resp.success else topic_id
    return state.topic_name_cache[topic_id]


def handle_chat(state: BotState, zenzap: ZenzapClient, openai_client: OpenAI, model: str, system_prompt: str, msg: dict) -> None:
    topic_id = msg["topicId"]
    text = msg.get("text", "")
    sender_name = msg.get("senderName", "Unknown")
    topic_name = resolve_topic_name(state, zenzap, topic_id)

    history = state.conversation_histories.setdefault(topic_id, [])
    history.append({"role": "user", "content": f"[from: {sender_name}, in: #{topic_name}]\n{text}"})

    # Keep the last 20 messages per topic to stay within token limits
    if len(history) > 20:
        history[:] = history[-20:]

    try:
        completion = openai_client.chat.completions.create(
            model=model,
            messages=[{"role": "system", "content": system_prompt}] + history,
        )
        reply = completion.choices[0].message.content
    except Exception as e:
        zenzap.send_message(topic_id, f"⚠️ OpenAI error: {e}")
        return

    history.append({"role": "assistant", "content": reply})
    zenzap.send_message(topic_id, reply)

Step 3 — Process Updates From the Poll Loop

Add a function that filters incoming updates and routes relevant ones to the chat handler. The bot should ignore its own messages and empty texts:

def handle_update(state: BotState, zenzap: ZenzapClient, openai_client: OpenAI, model: str, system_prompt: str, update: dict) -> None:
    if update.get("eventType") != "message.created":
        return

    msg = update.get("data", {}).get("message", {})
    if not msg:
        return

    if msg.get("senderId") == state.bot_member_id:
        return

    text = msg.get("text") or ""
    if not text.strip():
        return

    message_id = msg["id"]
    zenzap.mark_message_read(message_id)
    zenzap.add_reaction(message_id, "👀")

    handle_chat(state, zenzap, openai_client, model, system_prompt, msg)

Key behaviors:

Only message.created events are processed — see Webhook Events for all event types.
The bot skips its own messages by comparing senderId to bot_member_id.
Each incoming message is marked as read and given an 👀 reaction as visual feedback.

Step 4 — Wire Up the Main Loop

Finally, add the main() function that initializes the clients and starts the long-polling loop:

def main() -> None:
    load_dotenv()

    bot_api_key = os.getenv("BOT_API_KEY")
    bot_secret = os.getenv("BOT_SECRET")
    control_channel_id = os.getenv("CONTROL_CHANNEL_ID")
    openai_api_key = os.getenv("OPENAI_API_KEY")
    api_base_url = os.getenv("API_BASE_URL", "https://api.zenzap.co")
    openai_model = os.getenv("OPENAI_MODEL", "gpt-4o")

    for name, value in [
        ("BOT_API_KEY", bot_api_key),
        ("BOT_SECRET", bot_secret),
        ("CONTROL_CHANNEL_ID", control_channel_id),
        ("OPENAI_API_KEY", openai_api_key),
    ]:
        if not value:
            print(f"Missing required environment variable: {name}")
            sys.exit(1)

    zenzap = ZenzapClient(bot_api_key, bot_secret, api_base_url)
    openai_client = OpenAI(api_key=openai_api_key)

    # Verify the bot identity
    me_resp = zenzap.get_current_member()
    if not me_resp.success:
        print(f"Failed to fetch bot identity: {me_resp.data}")
        sys.exit(1)

    bot_member_id = me_resp.data["id"]
    bot_name = me_resp.data.get("name", "Bot")

    # Verify the control channel is accessible
    control_topic_resp = zenzap.get_topic(control_channel_id)
    if not control_topic_resp.success:
        print(f"Failed to fetch control channel: {control_topic_resp.data}")
        sys.exit(1)

    control_topic_name = control_topic_resp.data.get("name", control_channel_id)

    state = BotState(bot_member_id=bot_member_id)
    state.topic_name_cache[control_channel_id] = control_topic_name

    # Graceful shutdown
    def shutdown(signum, frame) -> None:
        state.running = False
        zenzap.send_message(control_channel_id, "🛑 Bot disconnecting...")
        sys.exit(0)

    signal.signal(signal.SIGINT, shutdown)
    signal.signal(signal.SIGTERM, shutdown)

    zenzap.send_message(
        control_channel_id,
        f"Agent connected successfully, this is the control channel #{control_topic_name}",
    )
    print(f"{bot_name} connected. Listening for messages...")

    # Long-polling loop
    while state.running:
        resp = zenzap.get_updates(offset=state.next_offset, poll_timeout=30)

        if not resp.success:
            if resp.status == 409:
                state.next_offset = None
            else:
                print(f"Poll error ({resp.status}): {resp.data}")
                time.sleep(2)
            continue

        state.next_offset = resp.data.get("nextOffset", state.next_offset)

        for update in resp.data.get("updates", []):
            handle_update(state, zenzap, openai_client, openai_model, DEFAULT_SYSTEM_PROMPT, update)


if __name__ == "__main__":
    main()

The main loop uses long polling to efficiently wait for new events. If a 409 error occurs (offset expired), it resets and starts fresh.

Step 5 — Run the Agent

Start the bot:

python bot.py

You should see output confirming the connection:

Bot connected. Listening for messages...

The bot will also send a message to your control channel confirming it’s online. From here, any message sent in a topic the bot has access to will trigger an OpenAI-powered response.

Next Steps

Custom tools — Extend handle_chat to support function calling so your agent can create tasks, manage topics, or call external APIs.
Conversation management — Use a database instead of in-memory conversation_histories for persistence across restarts.

Contact Support

If you have any questions or need help, don’t hesitate to contact our Support Team.

Get Started

Enterprise

FAQ

Guides

Integrations

Build Your Own Agent

Prerequisites

Project Setup

Step 1 — Build the Zenzap API Client

Step 2 — Handle Incoming Messages

Step 3 — Process Updates From the Poll Loop

Step 4 — Wire Up the Main Loop

Step 5 — Run the Agent

Next Steps

Contact Support

​Prerequisites

​Project Setup

​Step 1 — Build the Zenzap API Client

​Step 2 — Handle Incoming Messages

​Step 3 — Process Updates From the Poll Loop

​Step 4 — Wire Up the Main Loop

​Step 5 — Run the Agent

​Next Steps

​Contact Support

Prerequisites

Project Setup

Step 1 — Build the Zenzap API Client

Step 2 — Handle Incoming Messages

Step 3 — Process Updates From the Poll Loop

Step 4 — Wire Up the Main Loop

Step 5 — Run the Agent

Next Steps

Contact Support