Steeper

Telegram bot middleware that syncs incoming user messages and outgoing bot replies with the Steeper platform.

TL;DR: steeper is a thin middleware that plugs into any Telegram bot (aiogram / telebot / python-telegram-bot) and mirrors the entire conversation — incoming updates and the bot's outgoing replies — to the Steeper backend over HTTP. The backend stores the conversation, builds CRM/analytics, and streams real-time events to an operator panel.

Contents

• What is Steeper
• Installation
• Configuration
• Usage
• How it works
• Architecture
• The Steeper backend
• How they interact
• System logs
• Product events and funnels
• Library guarantees and behavior
• Backend compatibility
• License

What is Steeper

Steeper is a platform for working with Telegram bot conversations: a single place where you can see all user–bot dialogue, reply on behalf of the bot, run CRM and broadcasts, and view analytics.

For the platform to "see" a bot's traffic, the bot doesn't need to be rewritten: you just plug in the steeper library, which intercepts messages at the framework level and forwards them to the backend.

The ecosystem consists of three parts:

This repository is only the steeper library. The backend and panel live in their own repositories; they are described here only as much as needed to understand the integration.

Glossary

• bot_id — the bot's UUID, issued by the platform when the bot is registered.
• bot_token — the raw bot token from BotFather.
• token_hash — SHA-256(bot_token) in hex. The authentication secret: for both endpoints it is sent in the x-telegram-bot-api-secret-token header and never appears in the URL. The raw token is never sent over the network.
• Update — the standard Telegram Update object (as in the Bot API).
• Chat / Message — the platform's internal domain entities (with their own UUIDs) that Telegram traffic is turned into.

Installation

# Core (pick one extra for your framework)
pip install steeper[aiogram]     # aiogram v3
pip install steeper[telebot]     # pyTelegramBotAPI
pip install steeper[ptb]         # python-telegram-bot v20+

Need a backend? Steeper is self-hosted. Run the Steeper backend (Docker Compose), create a superuser, and register a bot to get its bot_id. Point base_url at your instance. See KarimovMurodilla/steeper-sdk for the backend and its self-hosting guide.

Runnable examples for every framework live in examples/.

Configuration

Every integration requires three values:

An optional timeout (seconds, default 10.0) is also accepted.

Optional: system logs

Set capture_logs=True and the middleware also ships the bot process's logging output to Steeper, where the operator panel shows it as a live stream with searchable history:

steeper = SteeperMiddleware(
    base_url="http://localhost:8000",
    bot_id="your-bot-uuid",
    bot_token=BOT_TOKEN,
    capture_logs=True,
    log_level="INFO",
)

steeper's own logger and its HTTP stack (httpx, httpcore, h11, …) are always excluded — shipping them would log from inside the shipping path, which is a loop that ends in a crash, not a dropped record. Anything you pass to log_exclude_loggers is added to that set, never replaces it.

Capture is off by default: it is extra outbound traffic and extra storage on the backend, so it should be an explicit choice.

Optional: event batching

track() needs no switch — it is off until the bot calls it, and the tracker opens no thread or connection before the first event. Two knobs tune the batching:

Prerequisite: register the bot

1. Bring up the Steeper backend (Docker Compose) and create a superuser.
2. Register the bot in the platform — you'll get its bot_id (UUID). The backend stores the bot's token_hash for authentication.
3. In the bot's code, pass base_url, bot_id, and bot_token to SteeperMiddleware.

Usage

aiogram v3

import asyncio

from aiogram import Bot, Dispatcher, Router
from aiogram.filters import CommandStart
from aiogram.types import Message
from steeper.integrations.aiogram import SteeperMiddleware

BOT_TOKEN = "123456:ABC-DEF..."

router = Router()


@router.message(CommandStart())
async def cmd_start(message: Message) -> None:
    await message.answer("Hello!")


async def main() -> None:
    bot = Bot(token=BOT_TOKEN)
    dp = Dispatcher()
    dp.include_router(router)

    steeper = SteeperMiddleware(
        base_url="http://localhost:8000",
        bot_id="your-bot-uuid",
        bot_token=BOT_TOKEN,
    )
    steeper.setup(dp, bot)

    await dp.start_polling(bot)


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

pyTelegramBotAPI (telebot)

import telebot
from steeper.integrations.telebot import SteeperMiddleware

BOT_TOKEN = "123456:ABC-DEF..."
bot = telebot.TeleBot(BOT_TOKEN)

steeper = SteeperMiddleware(
    base_url="http://localhost:8000",
    bot_id="your-bot-uuid",
    bot_token=BOT_TOKEN,
)
steeper.setup(bot)

# ... register your handlers as usual ...
bot.polling()

python-telegram-bot v20+

from telegram.ext import ApplicationBuilder
from steeper.integrations.ptb import SteeperMiddleware

BOT_TOKEN = "123456:ABC-DEF..."
app = ApplicationBuilder().token(BOT_TOKEN).build()

steeper = SteeperMiddleware(
    base_url="http://localhost:8000",
    bot_id="your-bot-uuid",
    bot_token=BOT_TOKEN,
)
steeper.setup(app)

# ... register your handlers as usual ...
app.run_polling()

How it works

All HTTP calls to Steeper go through SteeperRepository (steeper.repository): it forwards incoming updates and records outgoing bot messages to your backend. Log capture and event tracking are separate — see System logs and Product events and funnels. Each SteeperMiddleware exposes .repository (and .client for the underlying async HTTP client).

1. Incoming — the integration passes the full Telegram update, as Telegram-shaped JSON, to repository.forward_update(...) (every update type — messages, callback queries, inline queries, etc. — with all fields preserved). Your handlers still run as usual.

2. Outgoing — the integration hooks the framework so bot-originated messages are turned into OutgoingMessageSnapshot values and sent with repository.record_outgoing(...).

   • aiogram — Bot.__call__ is wrapped so any API call whose result is a Message (or a list of them, e.g. media groups) is logged—not only send_message.
   • python-telegram-bot — Bot._post is wrapped so JSON responses that decode to Message instances are logged (sends, edits, media groups, etc.).
   • telebot — telebot.apihelper._make_request is wrapped for your bot token so JSON result payloads that contain full message objects are logged.

If you bypass the normal API (e.g. raw HTTP to Telegram), call the repository yourself:

from steeper.repository import OutgoingMessageSnapshot

await steeper.repository.record_outgoing(
    OutgoingMessageSnapshot(
        chat_id=chat_id,
        message_id=message_id,
        text="visible text or caption",
        date=None,  # optional Unix ts; if omitted, the client defaults it to the current time
    )
)

All forwarding is fire-and-forget — no framework awaits the Steeper round-trip inline, so a slow or unreachable backend never adds latency to your handlers or to the bot's own API calls. Only the transport differs: for aiogram and python-telegram-bot the coroutine is scheduled on the running event loop; for telebot, whose handlers run on plain worker threads, it is scheduled on a shared background loop in a daemon thread. A failing backend never breaks the bot — see Library guarantees and behavior.

System logs

With capture_logs=True, setup() attaches a SteeperLogHandler to the root logger. It behaves differently from update forwarding, because log volume is orders of magnitude higher:

• Batched, not per-record. emit() only appends to an in-memory deque. A background thread flushes every log_flush_interval seconds, or as soon as log_batch_size records are waiting, and posts the batch in one request.
• Bounded, dropping the oldest. At most 10 000 records are held while the backend is unreachable. Past that the oldest are discarded — the opposite of the update forwarder, which drops the newest, because for logs the recent records are the ones being read.
• Never re-entrant. Records from steeper and its HTTP stack are dropped, and a thread-local guard breaks any remaining cycle, so a log emitted from inside the shipping path can't spiral.
• Never fatal. A record that cannot be formatted or serialized is skipped; logger.info(...) in your handlers never raises because of Steeper.
• Shipped on its own loop and client. Log batches go out on the shared background loop with a dedicated httpx.AsyncClient, so they never share a connection pool across event loops with update forwarding.

close() / aclose() detaches the handler and flushes what is still buffered — the one place the caller does wait for the network. For async frameworks the flush runs in a worker thread so it never blocks the bot's event loop.

To attach the handler somewhere other than the root logger, use it directly:

import logging
from steeper import SteeperLogHandler

handler = SteeperLogHandler(steeper.repository.config, level=logging.WARNING)
logging.getLogger("app.payments").addHandler(handler)

Product events and funnels

Telegram traffic tells the platform that a user sent a message. It cannot tell it that the user finished onboarding, opened a plan, or paid — those steps live inside your bot's code. track() is how they get out, and the platform builds funnels from them:

@router.message(F.text == "/buy")
async def buy(message: Message, tracker: EventTracker) -> None:
    tracker.track("checkout_started", user_id=message.from_user.id)
    ...
    tracker.track(
        "payment_succeeded",
        user_id=message.from_user.id,
        props={"plan": "pro", "amount": 4900},
    )

track() is a plain method, not a coroutine, on purpose: it only appends to a buffer, and making it awaitable would suggest the bot should wait for something. Call it the same way from an aiogram handler and a telebot worker thread.

An event that fails validation — an empty name, a user_id that is not an int — is dropped with a warning rather than raised. track() is called from inside handlers, and a mistyped analytics call must not take a reply down with it. The first rejection logs at WARNING and the rest at DEBUG, so a bad call in a hot handler cannot flood the output.

What your handlers should depend on

Give handlers middleware.tracker, not the middleware.

SteeperMiddleware owns an HTTP client, a repository, log capture, a patched Bot.__call__ and a lifecycle. A handler needs exactly one of those things — track — and a middleware is meant to be registered with the framework and then forgotten. EventTracker is that narrow dependency: it knows nothing about Telegram, or about which framework you use, so it drops into whatever wiring you already have.

async def main() -> None:
    steeper = SteeperMiddleware(base_url=..., bot_id=..., bot_token=BOT_TOKEN)

    bot = Bot(token=BOT_TOKEN)
    dp = Dispatcher(tracker=steeper.tracker)   # aiogram injects by parameter name
    dp.include_router(router)

    steeper.setup(dp, bot)   # after this line the middleware is not referred to again
    await dp.start_polling(bot)

Each framework has its own place for this, and the examples use it:

The payoff is that a handler becomes testable on its own, with no HTTP client, no dispatcher and no setup():

async def test_buy_reports_checkout() -> None:
    tracker = FakeTracker()
    await buy(message, tracker)
    assert tracker.events == [("checkout_started", 42)]

middleware.track(...) does the same thing and stays supported — it is the shorthand for a bot small enough not to want the wiring.

Behavior

Event shipping follows the log handler's shape, with one deliberate difference:

• Batched, not per-event. track() only appends to an in-memory deque. A background thread flushes every event_flush_interval seconds, or as soon as event_batch_size events are waiting, and posts the batch in one request.
• Timestamped when called, not when flushed. The backend orders funnel steps by ts, so stamping at flush time would collapse a batch into one instant and scramble every step order inside it.
• Bounded, dropping the oldest. At most 10 000 events are held while the backend is unreachable. Past that the oldest are discarded — the same choice as logs, for a different reason. Losing a funnel's entry events removes those users from the report entirely, which shrinks the window but leaves the conversion rate roughly unbiased; losing the later events of users already counted at step one drags the numerator down and reports conversion that never dropped as a drop. Dropping the oldest sheds whole users; dropping the newest would manufacture false non-conversion.
• Loud about failures. A failed batch logs at WARNING, unlike a log batch, which logs at DEBUG. A lost event is a permanently wrong funnel number and nothing else will ever mention it; log batches stay quiet only because warning about them would flood the very output being read.
• Flushed on shutdown. close() / aclose() ships what is still buffered before returning — the events held at that moment are the newest ones, which is exactly where a funnel's last step tends to live.
• Shipped on its own loop and client. Like log batches, events go out on the shared background loop with a dedicated httpx.AsyncClient.

Outside a middleware entirely — a worker, a management command, a web handler that shares the bot's backend — build a tracker of your own:

from steeper import EventTracker, SteeperConfig

tracker = EventTracker(SteeperConfig(base_url=..., bot_id=..., bot_token=...))
tracker.track("payment_succeeded", user_id=123456789)
...
tracker.close()   # ships what is still buffered

One tracker per process is the intent: each owns a buffer, a flush thread and a connection. Prefer passing steeper.tracker around over building a second one.

Shutting down

The middleware owns an httpx.AsyncClient, which should be closed when the bot stops. Where the framework offers a shutdown hook, setup() wires it up for you:

# telebot
try:
    bot.polling()
finally:
    steeper.close()

If you drive the lifecycle by hand — an aiogram dispatcher you feed yourself, or a PTB Application.shutdown() without run_polling (which does not run post_shutdown) — call await steeper.aclose() at the end.

Public API beyond the middleware

For manual scenarios, the package also exports:

• steeper.SteeperConfig — immutable config + validation, computes token_hash and the endpoint URLs.
• steeper.SteeperRepository — domain-oriented layer: forward_update(...), record_outgoing(...).
• steeper.SteeperClient — low-level async HTTP client (httpx).
• steeper.OutgoingMessageSnapshot — a normalized outgoing message.
• steeper.SteeperLogHandler — a logging.Handler that ships records to Steeper; steeper.install_log_handler(...) attaches one to a logger.
• steeper.EventTracker — buffers and ships product events; backs track(), and the dependency to hand to handlers. Reachable as middleware.tracker.

Internal layout

steeper/
├── _config.py        # SteeperConfig: validates base_url, token_hash, endpoint URLs
├── _client.py        # SteeperClient: httpx, sending, secret redaction in logs
├── _logging.py       # SteeperLogHandler: capture, batching, bounded buffer
├── _log_client.py    # SteeperLogClient: httpx client for the log endpoint
├── _events.py        # EventTracker: track(), batching, bounded buffer
├── _event_client.py  # SteeperEventClient: httpx client for the event endpoint
├── repository.py     # SteeperRepository + OutgoingMessageSnapshot
└── integrations/
    ├── aiogram.py     # SteeperMiddleware for aiogram v3
    ├── telebot.py     # SteeperMiddleware for pyTelegramBotAPI
    └── ptb.py         # SteeperMiddleware for python-telegram-bot v20+

Architecture

flowchart LR
    TG[Telegram] -->|update| BOT[Third-party bot\n+ steeper middleware]
    BOT -->|reply| TG

    subgraph CLIENT[Bot process]
        BOT --- LIB[steeper library]
    end

    LIB -->|"POST /v1/communications/webhook/{bot_id}"| API[Steeper Platform\nFastAPI]
    LIB -->|"POST /v1/communications/webhook/{bot_id}/bot-message"| API
    LIB -.->|"POST /v1/communications/webhook/{bot_id}/logs (optional)"| API
    LIB -.->|"POST /v1/communications/webhook/{bot_id}/events (optional)"| API

    API --> DB[(PostgreSQL)]
    API -->|publish| MQ{{RabbitMQ\nexchange: steeper.events}}
    MQ --> API
    API -->|WebSocket| UI[Operator panel]

The key idea: the library knows nothing about the platform's internal model. It talks to a handful of HTTP endpoints and passes Telegram traffic through in Telegram's own format. All domain logic — chats, users, funnel matching — is done by the backend.

The Steeper backend

The backend is the server side of the ecosystem and lives in its own repository:

➜ KarimovMurodilla/steeper-sdk

It is the FastAPI service this library talks to: it accepts incoming Telegram updates and outgoing bot messages, stores them verbatim, turns them into domain Chat / Message entities, maintains the Telegram-user CRM, publishes realtime events to the operator panel, and exposes the operator API (chat list, history, replies, analytics, broadcasts). It is self-hosted via Docker Compose.

Go there for deployment instructions, the full domain model, the realtime event contract, and the authoritative /v1 API reference. Everything this README says about the backend is only the slice needed to understand the integration; the endpoint contract itself is documented under How they interact.

How they interact

The HTTP contract

Every endpoint identifies the bot by bot_id in the path and authenticates with the secret (token_hash = SHA-256 of the bot token) in the x-telegram-bot-api-secret-token header — the secret never appears in the URL, and the raw bot_token is never sent over the network. Only the first two are always in play; logs and events are each used only when you turn them on.

A. Incoming update

POST {base_url}/v1/communications/webhook/{bot_id}
Header: x-telegram-bot-api-secret-token: <token_hash = SHA-256(bot_token)>
Body:   the full Telegram Update, as JSON (verbatim)

Backend responses: 200 (success), 400 (malformed payload), 403 (invalid secret), 404 (bot not found).

B. Outgoing bot message

POST {base_url}/v1/communications/webhook/{bot_id}/bot-message
Header: x-telegram-bot-api-secret-token: <token_hash = SHA-256(bot_token)>
Body:
{
  "chat_id":    123456789,        // Telegram chat id
  "text":       "visible text or caption",
  "message_id": 42,               // Telegram message id
  "date":       1700000000        // Unix ts; if omitted, the client sets the current time
}

Backend responses: 200, 400 (malformed payload), 403 (invalid secret), 404 (bot or Telegram user not found).

C. Log batch (only with capture_logs=True)

POST {base_url}/v1/communications/webhook/{bot_id}/logs
Header: x-telegram-bot-api-secret-token: <token_hash = SHA-256(bot_token)>
Body:
{
  "records": [                     // 1..500 records, oldest first
    {
      "ts":      1700000000.123,   // bot-side Unix timestamp
      "level":   "ERROR",          // DEBUG | INFO | WARNING | ERROR | CRITICAL
      "logger":  "app.handlers.start",
      "message": "Failed to answer callback query",
      "module":  "start",          // optional
      "func":    "cmd_start",      // optional
      "line":    42,               // optional
      "exc":     "Traceback ...",  // optional, formatted traceback
      "extra":   {"chat_id": 1}    // optional, structured context
    }
  ]
}

Backend responses: 200, 400 (malformed payload), 403 (invalid secret), 404 (bot not found), 500 (log storage unavailable). Every non-2xx is logged at DEBUG and the batch is dropped — the bot is never affected.

Incoming flow (user → bot → Steeper)

sequenceDiagram
    participant TG as Telegram
    participant Bot as Bot (+ steeper)
    participant API as Steeper backend
    participant DB as PostgreSQL
    participant MQ as RabbitMQ
    participant UI as Panel (WS)

    TG->>Bot: Update
    Note over Bot: steeper middleware<br/>runs BEFORE handlers
    Bot->>API: POST /webhook/{bot_id}<br/>+ secret header, raw Update
    Bot->>Bot: your handlers run as usual
    API->>API: verify bot_id + token_hash
    API->>DB: store raw update (idempotent by bot_id+update_id)
    alt it's a message and the bot is active
        API->>DB: upsert Telegram user (CRM)
        API->>DB: get/create Chat, store Message (sender=user)
        API->>MQ: publish chat.created (if the chat is new)
        API->>MQ: publish chat.message.created
        MQ-->>UI: event over WebSocket
    end
    API-->>Bot: 200 {success: true}

Backend specifics:

• Verbatim storage and idempotency. Every update is stored in full (even types not yet handled). The write is idempotent by (bot_id, update_id), so Telegram retries don't create duplicates.
• Only message / edited_message with a sender are turned into a domain chat. Everything else is simply logged.
• Inactive bot: the update is stored, but the chat workflow does not run.

Outgoing flow (bot replied → Steeper)

sequenceDiagram
    participant Bot as Bot (+ steeper)
    participant TG as Telegram
    participant API as Steeper backend
    participant DB as PostgreSQL

    Bot->>TG: send_message / reply (any API call)
    Note over Bot: steeper intercepts the Message result
    Bot->>API: POST /webhook/{bot_id}/bot-message<br/>+ secret header
    API->>API: resolve bot by bot_id, check token_hash, check active
    API->>DB: find Telegram user by chat_id
    API->>DB: get/create Chat, store Message (sender=bot)
    API-->>Bot: 200 {success: true}

Important notes about the outgoing flow:

• The bot-message endpoint stores the bot's message but, in the current implementation, does not publish a realtime event (unlike the incoming flow and replies sent by an operator from the panel).
• Logging an outgoing message requires that the Telegram user already exists (i.e. the dialogue usually had an incoming update first). Otherwise the backend responds 404, but that is not fatal for the bot (see below).

Library guarantees and behavior

• Never breaks the bot. If the backend is unreachable or returns an error, the library logs a warning and keeps going — your handlers and replies to the user are unaffected.
• Never slows the bot down. Every call to Steeper, log batches included, is fire-and-forget: the library schedules the request and returns immediately, so the client timeout (10s by default) bounds the background request, never your handler.
• At-most-once delivery. A failed forward is logged and dropped — there is no retry or persistent queue. Steeper is an observability sidecar, not a durable log: if the backend is down, that traffic is not recorded.
• Bounded memory. At most 512 forwards may be in flight at once. Past that the newest ones are dropped rather than queued, so a backend outage can't grow the bot's memory without limit. The first drop logs a warning; the rest log at debug with a running total, and the warning re-arms once the queue drains.
• Idempotent setup. Calling setup() twice on the same dispatcher/bot is a no-op, so an accidental double registration won't mirror every message twice.
• Safe logs. The token_hash is stripped from error text before logging (so the secret can't leak via a URL in an httpx message).
• Plaintext warning. If base_url is http:// against a non-local host, the library warns loudly: content and the secret would travel unencrypted. Use https:// in production.

Backend compatibility

This library talks to the Steeper backend's /v1 HTTP API. The endpoint contract above must match on the client and the server.

capture_logs=True additionally requires a backend that serves POST /v1/communications/webhook/{bot_id}/logs. Against an older backend the endpoint answers 404, log batches are dropped with a DEBUG line, and everything else keeps working.

track() likewise requires a backend serving POST /v1/communications/webhook/{bot_id}/events. An older one answers 404 and every batch is dropped — with a WARNING, not a DEBUG line, because unlike logs there is no other place the loss would show up.

As long as the backend keeps the v1 contract above, any 0.x client works. Breaking changes to the contract will bump the API version (/v2) and the library minor version together.

License

MIT — see LICENSE.