byteforge-telegram

A generic, reusable Python library for Telegram bot notifications and webhook management.

Features

• TelegramBotController: Send notifications via Telegram Bot API

  • Plain text messages
  • Formatted messages with title, fields, and footer
  • Rich Messages (Bot API 10.1): tables, lists, headings, formulas, media — see docs/rich-messages.md
  • Inline keyboards: attach buttons to sends, answer callback queries, edit sent messages
  • Both sync and async support
  • Automatic event loop handling
  • Session cleanup to prevent leaks

• WebhookManager: Manage Telegram webhooks

  • Set webhook URL
  • Get webhook information
  • Delete webhook
  • CLI tool included

Installation

pip install byteforge-telegram

Or install from source:

git clone https://github.com/jmazzahacks/byteforge-telegram.git
cd byteforge-telegram
pip install -e .

Quick Start

Sending Notifications

from byteforge_telegram import TelegramBotController, ParseMode

# Initialize with your bot token
bot = TelegramBotController("YOUR_BOT_TOKEN")

# Send a simple message
bot.send_message_sync(
    text="Hello from byteforge-telegram!",
    chat_ids=["CHAT_ID_1", "CHAT_ID_2"]
)

# Send a formatted message
bot.send_formatted_sync(
    title="Deployment Complete",
    fields={
        "Environment": "production",
        "Version": "1.2.3",
        "Status": "Success"
    },
    chat_ids=["YOUR_CHAT_ID"],
    emoji="✅",
    footer="Deployed at 2025-01-03 12:00:00 UTC"
)

Sending to a Supergroup Topic

Telegram supergroups with topics enabled require a message_thread_id to post into a specific topic. Use send_to_chat (single-target) rather than send_message (fan-out) — a thread id only applies to one supergroup, so it can't compose with a mixed list of chat ids:

bot.send_to_chat_sync(
    chat_id="-1001234567890",   # the supergroup
    text="<b>New ticket filed</b>",
    message_thread_id=42,        # the topic within it
)

Inline Keyboards and Callback Queries

Attach an inline keyboard to a single-chat send, then react to button taps and edit the delivered message. send_to_chat_sync returns the sent message_id (or None on failure), which you store to edit the message later:

keyboard = {
    "inline_keyboard": [[
        {"text": "✅ Approve", "callback_data": "approve:abc123"},
        {"text": "✖ Reject", "callback_data": "reject:abc123"},
    ]]
}

message_id = bot.send_to_chat_sync(
    chat_id="YOUR_CHAT_ID",
    text="<b>Escalation:</b> agent needs approval",
    reply_markup=keyboard,
)

# Later, in your webhook handler for the callback_query update:
bot.answer_callback_query_sync(callback_query_id, text="Approved!")

# Rewrite the message so the buttons disappear and the outcome is shown
bot.edit_message_text_sync(
    chat_id="YOUR_CHAT_ID",
    message_id=message_id,
    text="<b>Escalation:</b> approved by Jason",
)

The reply_markup dict is passed to the Bot API untouched. If the text is long enough to be split into multiple messages, the keyboard attaches to the last chunk, and the returned message_id is that last chunk's — so it is always the right target for edit_message_text_sync. Editing without reply_markup removes any existing keyboard; pass the keyboard again to keep it.

Sending a Rich Message

Rich Messages (Bot API 10.1) support structured content — headings, lists, tables, formulas, media, collapsible blocks — expressed as an extended-HTML or Markdown string. Use an InputRichMessage with send_rich_message / send_rich_message_sync:

from byteforge_telegram import InputRichMessage

bot.send_rich_message_sync(
    chat_id="123456789",
    rich_message=InputRichMessage(html=(
        "<h2>Daily report</h2>"
        "<ul><li>All systems green</li><li>3 deploys</li></ul>"
        "<table><tr><th>Metric</th><th>Value</th></tr>"
        "<tr><td>Uptime</td><td>99.98%</td></tr></table>"
    )),
)

Pass exactly one of html or markdown. Unlike send_message, rich text is sent as-is (no escaping/repair/splitting), so escape literal <, >, & yourself. See docs/rich-messages.md for the full list of supported tags, attributes, entities, and limits.

Managing Webhooks

Programmatic API

from byteforge_telegram import WebhookManager

# Initialize manager
manager = WebhookManager("YOUR_BOT_TOKEN")

# Set webhook
result = manager.set_webhook("https://example.com/telegram/webhook")
if result['success']:
    print(f"Webhook set: {result['description']}")

# Get webhook info
info = manager.get_webhook_info()
if info:
    print(f"Current webhook: {info.get('url')}")
    print(f"Pending updates: {info.get('pending_update_count')}")

# Delete webhook
result = manager.delete_webhook()
if result['success']:
    print("Webhook deleted")

Command-Line Interface

The package includes a setup-telegram-webhook CLI tool:

# Set webhook
setup-telegram-webhook --token YOUR_BOT_TOKEN --url https://example.com/telegram/webhook

# Or use environment variable
export TELEGRAM_BOT_TOKEN=YOUR_BOT_TOKEN
setup-telegram-webhook --url https://example.com/telegram/webhook

# Get webhook info
setup-telegram-webhook --token YOUR_BOT_TOKEN --info

# Delete webhook
setup-telegram-webhook --token YOUR_BOT_TOKEN --delete

API Reference

TelegramBotController

Methods

send_message_sync(text, chat_ids, parse_mode=ParseMode.HTML, ...)

• Send a plain text message (synchronous)
• Returns: Dict[str, bool] - success status for each chat

send_to_chat_sync(chat_id, text, *, message_thread_id=None, reply_markup=None, ...)

• Send a message to a single chat, optionally targeting a supergroup topic
• Returns: Optional[int] - the sent message's message_id, or None if the send did not fully succeed (truthy on success, so existing boolean-style checks keep working; on a multi-chunk send, None can mean earlier chunks were already delivered, so retrying may duplicate them)
• reply_markup takes a Bot API dict, e.g. {"inline_keyboard": [[{"text": ..., "callback_data": ...}]]}; when the text is split into chunks it attaches to the last chunk, whose message_id is the one returned
• Use this instead of send_message_sync when you need message_thread_id, since a thread id is only meaningful for one specific supergroup.

send_formatted_sync(title, fields, chat_ids, emoji=None, footer=None)

• Send a formatted message with title, fields, and footer (synchronous)
• Returns: Dict[str, bool] - success status for each chat

send_rich_message_sync(chat_id, rich_message, *, message_thread_id=None, disable_notification=False, protect_content=False)

• Send a Rich Message (Bot API 10.1) to a single chat; rich_message is an InputRichMessage
• Returns: bool - success status
• Content is sent as-is (no escaping/repair/splitting). See docs/rich-messages.md

edit_message_text_sync(chat_id, message_id, text, *, parse_mode=ParseMode.HTML, reply_markup=None, ...)

• Edit the text (and inline keyboard) of a previously sent message
• Returns: bool - success status
• Omitting reply_markup removes any existing keyboard; text is not split, so it must fit in one message (4096 chars)

answer_callback_query_sync(callback_query_id, *, text=None, show_alert=False)

• Answer an inline keyboard button tap (clears the spinner Telegram shows on the button)
• Returns: bool - success status
• text appears as a toast, or a modal alert with show_alert=True

send_message(...) / send_to_chat(...) / send_formatted(...) / send_rich_message(...) / edit_message_text(...) / answer_callback_query(...)

• Async versions of the above methods
• Use with await in async contexts

test_connection_sync(chat_id)

• Send a test message to verify bot is working
• Returns: bool

Parse Modes

from byteforge_telegram import ParseMode

ParseMode.HTML         # HTML formatting (default)
ParseMode.MARKDOWN     # Markdown formatting
ParseMode.MARKDOWN_V2  # MarkdownV2 formatting
ParseMode.NONE         # Plain text, no formatting

WebhookManager

Methods

set_webhook(webhook_url, timeout=10)

• Set the webhook URL for the bot
• Args:
  • webhook_url: HTTPS URL (required)
  • timeout: Request timeout in seconds
• Returns: Dict[str, Any] with success and description
• Raises: ValueError if URL is not HTTPS

get_webhook_info(timeout=10)

• Get current webhook configuration
• Returns: Dict[str, Any] with webhook details, or None on error

delete_webhook(timeout=10)

• Delete the current webhook
• Returns: Dict[str, Any] with success and description

TelegramResponse

Type-safe dataclass for constructing webhook responses.

Fields

• method: API method name (usually "sendMessage")
• chat_id: Target chat ID
• text: Message text
• parse_mode: Format type (default: "HTML")
• reply_markup: Optional keyboard markup
• disable_web_page_preview: Disable link previews (default: False)
• disable_notification: Send silently (default: False)

Methods

to_dict()

• Convert to JSON-serializable dictionary
• Returns: Dict[str, Any]

Example

from byteforge_telegram import TelegramResponse

response = TelegramResponse(
    method='sendMessage',
    chat_id=12345,
    text='<b>Hello!</b>',
    parse_mode='HTML',
    disable_web_page_preview=True
)

# Use in Flask webhook
return jsonify(response.to_dict()), 200

Examples

Integration with Flask (Simple)

import os
from flask import Flask, request, jsonify
from byteforge_telegram import TelegramBotController

app = Flask(__name__)
bot = TelegramBotController(os.getenv('TELEGRAM_BOT_TOKEN'))

@app.route('/telegram/webhook', methods=['POST'])
def telegram_webhook():
    update = request.get_json()

    # Process the update
    message = update.get('message', {})
    text = message.get('text', '')
    chat_id = str(message.get('chat', {}).get('id'))

    if text == '/start':
        bot.send_message_sync(
            text="Welcome! I'm your bot.",
            chat_ids=[chat_id]
        )

    return jsonify({'ok': True}), 200

Integration with Flask (Using TelegramResponse)

For more complex webhooks, use TelegramResponse for type-safe responses:

from flask import Flask, request, jsonify
from byteforge_telegram import TelegramResponse

app = Flask(__name__)

@app.route('/telegram/webhook', methods=['POST'])
def telegram_webhook():
    update = request.get_json()

    # Extract message details
    message = update.get('message', {})
    text = message.get('text', '')
    chat_id = message.get('chat', {}).get('id')

    # Handle command
    if text == '/start':
        response = TelegramResponse(
            method='sendMessage',
            chat_id=chat_id,
            text='<b>Welcome!</b> Type /help for commands.',
            parse_mode='HTML'
        )
        return jsonify(response.to_dict()), 200

    return jsonify({'ok': True}), 200

Async Usage

import asyncio
from byteforge_telegram import TelegramBotController, ParseMode

async def send_notifications():
    bot = TelegramBotController("YOUR_BOT_TOKEN")

    # Send multiple messages concurrently
    results = await bot.send_message(
        text="Async notification",
        chat_ids=["CHAT_1", "CHAT_2", "CHAT_3"],
        parse_mode=ParseMode.HTML
    )

    for chat_id, success in results.items():
        if success:
            print(f"Sent to {chat_id}")
        else:
            print(f"Failed to send to {chat_id}")

asyncio.run(send_notifications())

Error Handling

from byteforge_telegram import TelegramBotController

bot = TelegramBotController("YOUR_BOT_TOKEN")

results = bot.send_message_sync(
    text="Important notification",
    chat_ids=["CHAT_ID"]
)

for chat_id, success in results.items():
    if not success:
        print(f"Failed to send to {chat_id}")
        # Implement retry logic, logging, etc.

Design Philosophy

Sync/Async Compatibility

The library handles both synchronous and asynchronous contexts automatically:

• *_sync() methods work in regular Python code (like Flask apps)
• async methods work in async contexts (like FastAPI, async scripts)
• Automatically detects running event loops
• Creates fresh Bot instances per call to avoid loop conflicts

Session Management

Each message send creates a new Bot instance and properly cleans up the HTTP session afterward. This prevents connection leaks and event loop conflicts.

Error Handling

• Network errors are caught and logged
• Results dict shows success/failure per chat ID
• Graceful degradation when services are unavailable

Requirements

• Python 3.9+
• python-telegram-bot >= 20.0
• requests >= 2.31.0

Development

Setup

# Clone repository
git clone https://github.com/jmazzahacks/byteforge-telegram.git
cd byteforge-telegram

# Create and activate virtual environment
python3 -m venv .
source bin/activate

# Install development dependencies
pip install -r dev-requirements.txt

# Install package in development mode
pip install -e .

# Run tests
pytest

# Format code
black src/

Running Tests

# Run all tests
source bin/activate && pytest

# Run with coverage
source bin/activate && pytest --cov=byteforge_telegram

# Run specific test file
source bin/activate && pytest tests/test_models.py

License

MIT License - see LICENSE file for details

Contributing

Contributions are welcome! Please feel free to submit a Pull Request.

Author

Jason Byteforge (@jmazzahacks)

Links

• GitHub: https://github.com/jmazzahacks/byteforge-telegram
• Issues: https://github.com/jmazzahacks/byteforge-telegram/issues
• PyPI: https://pypi.org/project/byteforge-telegram/