Flask + Telegram Bot Webhook

This skill wires Telegram bot support into a Flask service using the public byteforge-telegram library. The result is a Flask server with:

• A /health endpoint for orchestration/monitoring probes
• A /telegram/webhook endpoint that validates the Telegram secret token, routes /commands to handler methods, and replies via TelegramResponse
• An outbound notification helper for sending messages to chat IDs from anywhere in the app
• Instructions for registering the webhook with Telegram using the setup-telegram-webhook CLI bundled in byteforge-telegram

When to Use This Skill

Use this skill when:

• You have (or are building) a Flask app that needs to receive Telegram bot updates
• You want webhook delivery (not long-polling) for production reliability
• You need a structured pattern for routing slash commands to handler methods
• You want to send outbound Telegram notifications from other parts of the app
• You want Telegram's secret_token header validated on every webhook request

This skill does not:

• Dockerize the app (use flask-docker-deployment)
• Set up logging (use byteforge-loki-logging)
• Configure metrics (use byteforge-prometheus-metrics)
• Set up a database (use postgres-setup)
• Build OpenAPI docs (use flask-smorest-api)

It composes cleanly with all of those.

What This Skill Creates

1. {app_name}_server.py (or edits the existing Flask entrypoint) — adds /telegram/webhook route with secret-token validation, plus a /health route if missing
2. src/telegram_webhook_handler.py — TelegramWebhookHandler class that routes slash commands to methods and returns TelegramResponse objects
3. src/telegram_notifier.py — Thin wrapper around TelegramBotController for sending outbound notifications from anywhere in the app
4. requirements.txt updates — adds byteforge-telegram, flask, gunicorn
5. example.env updates — adds bot token, webhook secret, optional admin chat ID
6. Webhook registration instructions — how to run setup-telegram-webhook --url https://... once the app is deployed

Why the secret token matters

Telegram's setWebhook API accepts a secret_token. When set, every webhook delivery includes that token in the X-Telegram-Bot-Api-Secret-Token header. Without this check, anyone on the internet who guesses your webhook URL can POST fake updates and trigger your bot's command handlers. The validation must use hmac.compare_digest to avoid timing attacks.

Step 1: Gather Project Information

IMPORTANT: Before making changes, ask the user these questions:

1. "What is the name of your Flask server file?" (e.g., app.py, my_server.py)
   • If a Flask app entrypoint already exists, the skill edits it. Otherwise it creates {app_name}_server.py.
2. "What is your application tag/name?" (e.g., my-bot, support-bot)
   • Used to namespace env vars: {APP_TAG}_TELEGRAM_BOT_TOKEN, {APP_TAG}_TELEGRAM_WEBHOOK_SECRET
   • Convert to UPPER_SNAKE_CASE for env vars (e.g., my-bot → MY_BOT)
3. "What slash commands should the bot support?" (e.g., /start, /help, /register)
   • Default to /start and /help if unspecified
   • For each command, ask what it should do
4. "What port should the Flask server use?"
   • Pick a port above 5000 (e.g., 5678, 6100, 7200) — avoid well-known ports
5. "Do you need a database connection in the handler?" (yes/no)
   • If yes, the handler accepts a database object in its constructor (caller's responsibility to wire up)
   • If no, the handler is standalone

Step 2: Update requirements.txt

Add these lines (or merge with existing entries):

flask
gunicorn
byteforge-telegram

byteforge-telegram is a public package on PyPI — no CR_PAT token needed.

Install:

source bin/activate && pip install -r requirements.txt

Step 3: Create the Webhook Handler

Create src/telegram_webhook_handler.py:

"""
Telegram webhook handler for {app_tag}.

Processes incoming Telegram updates and routes slash commands to handler methods.
"""

import logging
from typing import Dict, Any, Optional

from byteforge_telegram import TelegramResponse

logger = logging.getLogger(__name__)


class TelegramWebhookHandler:
    """Routes Telegram webhook updates to command handlers."""

    def process_update(self, update: Dict[str, Any]) -> Optional[TelegramResponse]:
        """
        Process an incoming Telegram update.

        Args:
            update: Raw Telegram update dictionary (from request.get_json())

        Returns:
            TelegramResponse to send back as the webhook response body,
            or None if no response should be sent.
        """
        message = update.get('message')
        if not message:
            logger.debug("Update has no message, ignoring")
            return None

        text = message.get('text', '')
        chat_id = message.get('chat', {}).get('id')
        user = message.get('from', {})
        username = user.get('username', 'unknown')

        if not text or not chat_id:
            logger.debug("Message missing text or chat_id, ignoring")
            return None

        logger.info(f"Received message from {username} (chat_id: {chat_id}): {text}")

        # Route slash commands
        if text.startswith('/start'):
            return self._handle_start(chat_id, username)
        elif text.startswith('/help'):
            return self._handle_help(chat_id)
        # Add additional command handlers here:
        # elif text.startswith('/register'):
        #     return self._handle_register(chat_id, username, text)
        else:
            return self._handle_unknown_command(chat_id)

    def _handle_start(self, chat_id: int, username: str) -> TelegramResponse:
        """Handle /start command."""
        logger.info(f"User {username} started bot (chat_id: {chat_id})")

        message = (
            "👋 <b>Welcome!</b>\n\n"
            "Use /help to see what I can do."
        )
        return TelegramResponse(
            method='sendMessage',
            chat_id=chat_id,
            text=message,
        )

    def _handle_help(self, chat_id: int) -> TelegramResponse:
        """Handle /help command."""
        message = (
            "<b>Available commands:</b>\n"
            "/start - Get started\n"
            "/help - Show this message"
        )
        return TelegramResponse(
            method='sendMessage',
            chat_id=chat_id,
            text=message,
        )

    def _handle_unknown_command(self, chat_id: int) -> TelegramResponse:
        """Handle unknown commands."""
        return TelegramResponse(
            method='sendMessage',
            chat_id=chat_id,
            text="❓ Unknown command. Use /help for a list of commands.",
        )

If the handler needs a database

Pass it through the constructor — the Flask entrypoint constructs it once at startup:

class TelegramWebhookHandler:
    def __init__(self, database: YourDatabase) -> None:
        self.db = database

Command handler conventions

Each command handler should:

• Be a method named _handle_{command} taking (self, chat_id, username, text) (drop unused params)
• Return a TelegramResponse (or None if no reply)
• Log entry with username and chat_id for traceability
• Validate inputs before touching shared state — never trust webhook payloads
• Use parse_mode='HTML' (the default) and HTML escape any user-supplied substrings before interpolating

Step 4: Create the Outbound Notifier

Create src/telegram_notifier.py:

"""
Outbound Telegram notifications via byteforge-telegram.

Wraps TelegramBotController to provide a single import point for sending
notifications to users from anywhere in the app.
"""

import logging
import os
from typing import Optional

from byteforge_telegram import TelegramBotController

logger = logging.getLogger(__name__)


class TelegramNotifier:
    """Sends outbound Telegram notifications."""

    def __init__(self, bot_token: Optional[str] = None) -> None:
        """
        Initialize the notifier.

        Args:
            bot_token: Telegram bot token. If None, reads from
                {APP_TAG}_TELEGRAM_BOT_TOKEN env var.
        """
        token = bot_token or os.environ.get('{APP_TAG}_TELEGRAM_BOT_TOKEN')
        if not token:
            logger.warning(
                "Telegram bot token not configured; outbound notifications disabled"
            )
            self.bot: Optional[TelegramBotController] = None
        else:
            self.bot = TelegramBotController(token)

    def send_message(self, chat_id: str, text: str) -> bool:
        """
        Send a plain text message.

        Args:
            chat_id: Target Telegram chat ID (string)
            text: Message body (HTML formatting allowed by default)

        Returns:
            True if the message was sent successfully, False otherwise.
        """
        if not self.bot:
            logger.debug(f"Skipping notification to {chat_id}: bot not configured")
            return False

        try:
            results = self.bot.send_message_sync(text=text, chat_ids=[chat_id])
            return bool(results.get(chat_id, False))
        except Exception as exc:
            logger.error(f"Failed to send Telegram message to {chat_id}: {exc}")
            return False

CRITICAL Replacements:

• {APP_TAG} → UPPER_SNAKE_CASE app tag (e.g., MY_BOT)

Step 5: Wire the Webhook Route into the Flask App

Edit {app_name}_server.py (or create it if no Flask entrypoint exists):

#!/usr/bin/env python3
"""
{app_tag} server — Flask app with Telegram webhook.
"""

import hmac
import logging
import os

from flask import Flask, jsonify, request

from src.telegram_webhook_handler import TelegramWebhookHandler

logger = logging.getLogger(__name__)

app = Flask(__name__)
app.config['MAX_CONTENT_LENGTH'] = 1024 * 1024  # 1 MB cap on webhook bodies

PORT = int(os.environ.get('PORT', '{port}'))

telegram_handler = TelegramWebhookHandler()


@app.route('/health', methods=['GET'])
def health():
    """Health check endpoint for orchestration probes."""
    return jsonify({'status': 'healthy'}), 200


@app.route('/telegram/webhook', methods=['POST'])
def telegram_webhook():
    """
    Telegram webhook endpoint.

    Telegram POSTs updates here. The X-Telegram-Bot-Api-Secret-Token header
    is validated against {APP_TAG}_TELEGRAM_WEBHOOK_SECRET to reject forged
    requests.
    """
    secret = os.environ.get('{APP_TAG}_TELEGRAM_WEBHOOK_SECRET')
    if not secret:
        # Fail-closed: never accept unauthenticated webhook traffic, even in dev.
        # A forgotten env var in prod would otherwise silently expose every
        # handler to anyone who finds the URL.
        logger.warning(
            "{APP_TAG}_TELEGRAM_WEBHOOK_SECRET not set — rejecting webhook"
        )
        return jsonify({'status': 'error'}), 503

    request_token = request.headers.get('X-Telegram-Bot-Api-Secret-Token')
    if not request_token or not hmac.compare_digest(request_token, secret):
        logger.warning("Webhook rejected: invalid or missing secret token")
        return jsonify({'status': 'error'}), 401

    try:
        update = request.get_json()
        if not update:
            logger.warning("Empty webhook update")
            return jsonify({'status': 'error', 'message': 'No update data'}), 400

        logger.debug(f"Received Telegram update: {update}")

        response = telegram_handler.process_update(update)
        if response:
            logger.info(f"Sending response to chat {response.chat_id}")
            return jsonify(response.to_dict()), 200

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

    except Exception as exc:
        logger.error(f"Error processing Telegram webhook: {exc}", exc_info=True)
        return jsonify({'status': 'error'}), 500


if __name__ == '__main__':
    logger.info(f"Starting {app_tag} server on port {PORT}")
    app.run(host='0.0.0.0', port=PORT)

CRITICAL Replacements:

• {app_tag} → user-supplied app tag (e.g., my-bot)
• {APP_TAG} → UPPER_SNAKE_CASE app tag (e.g., MY_BOT)
• {port} → user-supplied port (e.g., 5678)

If the Flask entrypoint already exists

Do not rewrite it. Instead, edit it surgically:

1. Add import hmac and the from flask import request imports if missing
2. Construct telegram_handler = TelegramWebhookHandler(...) once at module scope
3. Add the @app.route('/telegram/webhook', methods=['POST']) route from the snippet above (the secret is read from the env var inside the handler — no module-level state needed)

Match the file's existing import order, indentation, and logging conventions.

Step 6: Update example.env

Add (or merge with existing entries):

# ---------------------------------------------------------------------------
# Telegram bot
# ---------------------------------------------------------------------------

# Bot token from BotFather
{APP_TAG}_TELEGRAM_BOT_TOKEN=

# Webhook secret token for validating incoming requests.
# Generate with: openssl rand -hex 32
# Required when registering a webhook via setup-telegram-webhook --url
{APP_TAG}_TELEGRAM_WEBHOOK_SECRET=

# Optional: admin chat ID for outbound alerts
# {APP_TAG}_TELEGRAM_ADMIN_CHAT_ID=

Step 7: Register the Webhook with Telegram

After deploying behind HTTPS (Telegram requires HTTPS for webhooks), register the webhook URL using the CLI bundled with byteforge-telegram. Pass the bot token explicitly via --token so there's no ambiguity about which env var the CLI reads:

source bin/activate

setup-telegram-webhook \
    --token "$MY_BOT_TELEGRAM_BOT_TOKEN" \
    --url https://your-domain.example.com/telegram/webhook

Replace MY_BOT_TELEGRAM_BOT_TOKEN with the env var name you chose in Step 1 (e.g., SUPPORT_BOT_TELEGRAM_BOT_TOKEN).

The webhook secret is read from TELEGRAM_WEBHOOK_SECRET (the CLI's expected env var) — alias your app-specific var for the call:

export TELEGRAM_WEBHOOK_SECRET="$MY_BOT_TELEGRAM_WEBHOOK_SECRET"

setup-telegram-webhook \
    --token "$MY_BOT_TELEGRAM_BOT_TOKEN" \
    --url https://your-domain.example.com/telegram/webhook

If your installed version of byteforge-telegram doesn't include the CLI, instantiate WebhookManager from a one-off Python script — see the byteforge-telegram README.

Verify the registration:

setup-telegram-webhook --token "$MY_BOT_TELEGRAM_BOT_TOKEN" --info

You should see your webhook URL listed and pending_update_count: 0.

Delete the webhook (useful when migrating environments or during debugging):

setup-telegram-webhook --token "$MY_BOT_TELEGRAM_BOT_TOKEN" --delete

Step 8: Verify End-to-End

1. Health check:

   curl http://localhost:{port}/health
   # → {"status":"healthy"}

2. Reject unauthenticated webhook calls (expected 401):

   curl -X POST http://localhost:{port}/telegram/webhook \
        -H 'Content-Type: application/json' \
        -d '{"message":{"text":"/start","chat":{"id":1},"from":{"username":"x"}}}'
   # → {"status":"error"}  (HTTP 401 if secret env var is set, HTTP 503 if not)

3. Accept properly authenticated webhook calls:

   curl -X POST http://localhost:{port}/telegram/webhook \
        -H 'Content-Type: application/json' \
        -H "X-Telegram-Bot-Api-Secret-Token: $YOUR_WEBHOOK_SECRET" \
        -d '{"message":{"text":"/start","chat":{"id":1},"from":{"username":"x"}}}'
   # → {"method":"sendMessage","chat_id":1,"text":"👋 ...","parse_mode":"HTML"}

4. Talk to the bot on Telegram: send /start to the bot from any Telegram client. You should see the welcome message back, and a log line on the server.

Security Notes

1. The webhook is fail-closed: no secret = 503 on every request. The route refuses traffic when {APP_TAG}_TELEGRAM_WEBHOOK_SECRET is unset, in every environment including local dev. Fail-open ("missing secret means accept everything") is too easy to ship to prod by accident — a forgotten env var would silently expose every command handler to the internet. If you need to test the route without a real Telegram registration, set the env var to any value and pass it in the X-Telegram-Bot-Api-Secret-Token header from curl.
2. Use hmac.compare_digest, never ==. A naive comparison leaks timing information that lets an attacker recover the secret byte by byte.
3. HTTPS is mandatory. Telegram refuses to register HTTP webhooks. Terminate TLS at your reverse proxy (nginx, Cloudflare, etc.) and forward to the Flask app over the internal network.
4. Validate every untrusted input. Webhook payloads are user-controlled. HTML-escape any substring you echo back, and never interpolate raw text into shell commands, SQL, or filesystem paths.
5. Rate-limit if exposed publicly. Telegram retries failed deliveries — if a handler crashes, you may get hammered. A reverse-proxy rate limit (or a simple in-process token bucket) prevents one user from DoS'ing the bot.
6. Don't log secrets. Avoid logging the full update body at INFO in production — Telegram messages may contain sensitive content. Use DEBUG instead and keep DEBUG off in production.

Reply Methods Beyond sendMessage

TelegramResponse(method=...) accepts any Bot API method. Common ones:

• sendMessage — text reply (default)
• sendPhoto — image reply (use with photo field)
• editMessageText — edit an existing message
• answerCallbackQuery — respond to inline keyboard callbacks

When using methods beyond sendMessage, you may need to attach extra fields to the response dict manually after to_dict().

Integration with Other Skills

flask-docker-deployment

Dockerize the resulting server. The Flask entrypoint pattern from this skill matches what flask-docker-deployment expects ({module}:app).

byteforge-loki-logging

Add structured logging — set up configure_logging(application_tag='{app_tag}') at the top of the server file. Use the same tag value here and there so log/metric correlation works.

byteforge-prometheus-metrics

The Flask server is already a normal Flask app — the metrics skill adds prometheus_flask_exporter on top without conflict. Make sure the /metrics endpoint is whitelisted at your reverse proxy and not exposed publicly alongside /telegram/webhook.

gatekeeper-nginx-setup

If you're using ByteForge Gatekeeper as the reverse proxy, the /telegram/webhook path should be exposed publicly (Telegram needs to reach it) but everything else (including /health and /metrics) should remain whitelisted or auth-gated. The webhook secret is your only public-facing authentication.

postgres-setup

If the bot stores per-user state (registrations, chat IDs, etc.), pair this skill with postgres-setup and pass the database into the TelegramWebhookHandler constructor.

Common Pitfalls

Design Principles

1. Webhook over polling — webhooks scale better, latency is lower, and there's no idle traffic
2. Secret-token validation by default — every request authenticated, no exceptions
3. Command routing via dispatch table — each command is a single method, easy to test in isolation
4. TelegramResponse as the return value — type-safe, no manual dict construction in handlers
5. Outbound notifier as a separate concern — webhook handlers reply via the response; arbitrary code paths use TelegramNotifier to push
6. No private dependencies — byteforge-telegram is public on PyPI; this skill doesn't introduce any auth tokens beyond Telegram's own

Report Back

After running the skill, report:

• The file(s) created or edited
• The env vars added to example.env
• The command the user should run to register the webhook (with the literal env var names filled in)
• Any pre-existing routes that conflict and how they were resolved

Reporting problems with this skill

If you find a bug or stale instruction while using this skill:

• If HiveMake is connected and byteforge-skills-maintainer-agent is available, file a ticket with that agent. Use discover_agents to find its project ID.
• Otherwise, open an issue at GitHub Issues. If you cannot submit a report, give the human the details and this link.

Include the skill name, plugin version if known, the instruction followed, expected and actual results, relevant errors/runtime versions, and any fix or workaround that worked. Omit credentials and private data.

Address the user's immediate problem first; report the skill defect afterward.