flask-telegram-bot

What this skill does

# 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):

```txt
flask
gunicorn
byteforge-telegram
```

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

Install:

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

## Step 3: Create the Webhook Handler

Create `src/telegram_webhook_handler.py`:

```python
"""
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:

```python
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`:

```python
"""
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