steeper 0.1.2

Telegram bot middleware for Steeper — intercepts updates and outgoing messages to sync with the Steeper platform.

Steeper

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

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+

Configuration

Every integration requires three values:

Parameter	Description
base_url	Steeper backend URL (e.g. http://localhost:8000)
bot_id	UUID of the bot registered in Steeper
bot_token	Raw Telegram bot token from BotFather

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. Each SteeperMiddleware exposes .repository (and .client for the underlying async HTTP client).

1. Incoming — the integration passes each update as Telegram-shaped JSON to repository.forward_update(...). 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
    )
)

Failures are never fatal: if the Steeper backend is unreachable or returns an error, a warning is logged and your bot keeps working. Note the dispatch model differs per framework — for aiogram and python-telegram-bot the sync calls are awaited inline, so a slow or unreachable backend can add latency (up to the client timeout, 10s by default) per update; telebot dispatches them as background tasks.

License

MIT