Building a Practical Telegram Open Source Workflow: Self-Hosting a Bot Update Relay

Telegram’s open source ecosystem is most useful when you can run a real workflow end to end. In this tutorial, we’ll build a practical bot update relay that receives Telegram Bot API updates, validates them, stores them locally, and forwards them into a simple internal processing pipeline you can adapt for production.

Telegram is unusual among major messaging platforms because a large part of its ecosystem is open source, including the client apps and several official components. For developers, that matters less as a slogan and more as a practical advantage: you can inspect how parts of the platform behave, study protocol details, and build tooling around Telegram with fewer black boxes.

This tutorial focuses on one concrete implementation: a self-hosted bot update relay. The relay receives updates from Telegram, verifies that they are legitimate, stores them in a local queue, and forwards them into your internal application logic.

Why build this instead of wiring Telegram directly to your business logic?

You get a clean boundary between Telegram traffic and your app.
You can replay updates for debugging.
You can buffer spikes and avoid losing events during deployments.
You can attach multiple processors later without changing the bot webhook.

The goal here is not to build a toy bot. The goal is to show a production-shaped pattern that Telegram developers can actually reuse.

What we are building

We will create a small service with three parts:

Webhook receiver — accepts Telegram Bot API updates.
Validation and persistence layer — checks the payload and stores it.
Worker pipeline — processes updates asynchronously.

The implementation will use:

Python 3.11+
FastAPI for the webhook endpoint
SQLite for local persistence during development
Redis queue optional for production handoff
Telegram Bot API webhook mode

If you already use Node.js, Go, or Rust, the architectural pattern stays the same. Only the syntax changes.

Prerequisites

You need:

a Telegram bot token from BotFather
a publicly reachable HTTPS endpoint for webhooks
a basic understanding of JSON and async web services

For local development, use something like ngrok, cloudflared, or a similar tunneling tool. Telegram webhooks require HTTPS.

Project structure

Here is the layout we will use:

telegram-relay/
├── app/
│   ├── main.py
│   ├── models.py
│   ├── storage.py
│   ├── worker.py
│   └── settings.py
├── requirements.txt
└── README.md

Step 1: Define the update schema

Telegram sends updates as JSON objects. You do not need to model every field to begin with, but you should strongly type the parts you care about.

Create app/models.py:

from pydantic import BaseModel
from typing import Optional, List


class User(BaseModel):
    id: int
    is_bot: bool
    first_name: str
    username: Optional[str] = None


class Chat(BaseModel):
    id: int
    type: str
    title: Optional[str] = None
    username: Optional[str] = None


class Message(BaseModel):
    message_id: int
    date: int
    chat: Chat
    from_user: Optional[User] = None
    text: Optional[str] = None


class Update(BaseModel):
    update_id: int
    message: Optional[Message] = None

A few notes:

Telegram payloads contain many optional branches: message, edited_message, callback_query, inline_query, and more.
Start with only the branches your bot needs.
Use strict validation to reject malformed payloads early.

If you want to expand later, add more fields incrementally rather than copying the entire Telegram schema.

Step 2: Add configuration

Create app/settings.py:

import os


BOT_TOKEN = os.getenv("BOT_TOKEN", "")
WEBHOOK_SECRET = os.getenv("WEBHOOK_SECRET", "")
DATABASE_PATH = os.getenv("DATABASE_PATH", "./telegram_updates.db")

Use environment variables for secrets. Do not hardcode your bot token.

The webhook secret is optional but strongly recommended. Telegram supports a secret token header so your server can verify that a request came from Telegram and not from a random client.

Step 3: Create persistence

For development, SQLite is enough. It gives you a durable event log with very little setup.

Create app/storage.py:

import json
import sqlite3
from datetime import datetime
from app.settings import DATABASE_PATH


def init_db():
    conn = sqlite3.connect(DATABASE_PATH)
    cur = conn.cursor()
    cur.execute("""
        CREATE TABLE IF NOT EXISTS updates (
            update_id INTEGER PRIMARY KEY,
            payload TEXT NOT NULL,
            received_at TEXT NOT NULL,
            processed INTEGER NOT NULL DEFAULT 0
        )
    """)
    conn.commit()
    conn.close()


def save_update(update_id: int, payload: dict):
    conn = sqlite3.connect(DATABASE_PATH)
    cur = conn.cursor()
    cur.execute(
        "INSERT OR REPLACE INTO updates (update_id, payload, received_at, processed) VALUES (?, ?, ?, 0)",
        (update_id, json.dumps(payload), datetime.utcnow().isoformat())
    )
    conn.commit()
    conn.close()


def fetch_unprocessed(limit: int = 50):
    conn = sqlite3.connect(DATABASE_PATH)
    cur = conn.cursor()
    cur.execute(
        "SELECT update_id, payload FROM updates WHERE processed = 0 ORDER BY update_id ASC LIMIT ?",
        (limit,)
    )
    rows = cur.fetchall()
    conn.close()
    return rows


def mark_processed(update_id: int):
    conn = sqlite3.connect(DATABASE_PATH)
    cur = conn.cursor()
    cur.execute("UPDATE updates SET processed = 1 WHERE update_id = ?", (update_id,))
    conn.commit()
    conn.close()

This table lets you:

retain raw payloads for debugging
replay updates after a crash
inspect bot activity manually

For heavier traffic, swap SQLite for Postgres or enqueue updates into Redis or RabbitMQ.

Step 4: Build the webhook receiver

Create app/main.py:

from fastapi import FastAPI, Request, Header, HTTPException
from app.models import Update
from app.storage import init_db, save_update
from app.settings import WEBHOOK_SECRET

app = FastAPI()


@app.on_event("startup")
def startup_event():
    init_db()


@app.post("/telegram/webhook")
async def telegram_webhook(
    request: Request,
    x_telegram_bot_api_secret_token: str | None = Header(default=None)
):
    if WEBHOOK_SECRET and x_telegram_bot_api_secret_token != WEBHOOK_SECRET:
        raise HTTPException(status_code=403, detail="Invalid secret token")

    payload = await request.json()
    update = Update.model_validate(payload)

    # Persist raw payload before doing anything else.
    save_update(update.update_id, payload)

    return {"ok": True}

This is intentionally minimal but already useful.

Key implementation choices:

Validate before processing to avoid broken payloads entering your system.
Persist first, process later to reduce event loss.
Return quickly so Telegram does not retry due to slow responses.

Telegram may retry webhook deliveries if your endpoint is slow or returns errors. Fast acknowledgment helps avoid duplicates and backpressure.

Step 5: Add a worker loop

Now create app/worker.py:

import json
import time
from app.storage import fetch_unprocessed, mark_processed


def handle_update(update_id: int, payload: dict):
    message = payload.get("message")
    if not message:
        return

    text = message.get("text", "")
    chat_id = message["chat"]["id"]

    print(f"Processing message in chat {chat_id}: {text}")

    # Replace this with your business logic.
    # Example: enqueue a task, store a domain event, or send a reply.


def run_worker():
    while True:
        rows = fetch_unprocessed()
        if not rows:
            time.sleep(1)
            continue

        for update_id, payload_str in rows:
            payload = json.loads(payload_str)
            try:
                handle_update(update_id, payload)
                mark_processed(update_id)
            except Exception as exc:
                print(f"Failed to process update {update_id}: {exc}")


if __name__ == "__main__":
    run_worker()

This worker demonstrates the relay pattern: the webhook layer only receives and stores updates, while the worker performs the actual business logic.

That separation is the main practical value here. It keeps Telegram delivery concerns away from your core system.

Step 6: Register the webhook with Telegram

You can set the webhook using Bot API methods. Replace values as needed:

curl -X POST "https://api.telegram.org/bot$BOT_TOKEN/setWebhook" \
  -d "url=https://your-domain.example/telegram/webhook" \
  -d "secret_token=$WEBHOOK_SECRET"

To inspect the current webhook:

curl "https://api.telegram.org/bot$BOT_TOKEN/getWebhookInfo"

If you need to remove it during testing:

curl -X POST "https://api.telegram.org/bot$BOT_TOKEN/deleteWebhook"

When debugging webhook issues, getWebhookInfo is one of the first things you should check. It tells you whether Telegram is delivering, retrying, or failing.

Step 7: Test with a real chat

Start the API server:

uvicorn app.main:app --reload --port 8000

Start the worker in another terminal:

python -m app.worker

Then send a message to your bot in Telegram.

If everything is working, you should see:

the update stored in SQLite
the worker printing the message text
the row marked as processed

You can inspect the database manually:

sqlite3 telegram_updates.db

Then:

SELECT update_id, processed, received_at FROM updates ORDER BY update_id DESC;

Making it useful in a real Telegram codebase

At this point, you have a relay. The next step is turning it into a reusable platform component.

1. Separate Telegram transport from domain logic

Do not let your bot handlers become the application.

Instead:

parse Telegram updates at the edge
convert them into internal events
let your domain services consume those events

Example:

def handle_update(update_id: int, payload: dict):
    message = payload.get("message")
    if not message:
        return

    event = {
        "type": "telegram.message.received",
        "chat_id": message["chat"]["id"],
        "user_id": message.get("from", {}).get("id"),
        "text": message.get("text", ""),
        "update_id": update_id,
    }

    publish_event(event)

This is especially valuable if you have multiple bots or multiple entry points into the same app.

2. Store raw payloads, not just normalized data

Raw payloads help when Telegram adds new fields or when your code misbehaves.

You can always derive normalized records later. You cannot reconstruct missing data if you discard it too early.

3. Design for idempotency

Telegram can resend updates, and your service can restart in the middle of processing.

Use update_id as the natural deduplication key. In the sample code, update_id is the primary key in SQLite. That prevents duplicate inserts and gives you a basic idempotency guard.

4. Use the webhook secret

Without request verification, any public endpoint can be spammed with fake traffic.

The secret token header is a cheap and effective first line of defense.

5. Keep webhook latency low

If your processing takes more than a few hundred milliseconds, do not do it inline.

Acknowledge quickly, then hand off work to a queue or worker.

Production hardening checklist

If you plan to deploy this pattern beyond development, add these improvements:

Postgres instead of SQLite for multi-process or multi-instance deployments
Redis queue or background jobs for asynchronous processing
Structured logging with update IDs and chat IDs
Metrics for request rate, processing lag, and failure counts
Retry policy for transient downstream failures
Dead-letter storage for updates that cannot be processed
Rate limiting if you expose additional admin endpoints
Secret rotation for webhook verification

A particularly useful metric is processing lag: the difference between the update’s arrival time and the moment your worker finishes it. That number will tell you when your pipeline is falling behind.

Common mistakes Telegram developers make

Processing directly in the webhook handler

It works at first, then fails under load or during deployments. Keep the handler thin.

Ignoring update retries

Telegram is reliable, but your endpoint may not be. Assume duplicates will happen.

Modeling too much too early

Telegram payloads are broad. Only model the fields you actually consume.

Not preserving raw input

Once you normalize aggressively, debugging gets harder.

Treating bots as stateless scripts

For real apps, Telegram integrations are event systems, not one-off scripts.

Where open source helps in practice

Telegram’s open source footprint is helpful because it encourages a deeper engineering workflow:

you can inspect client behavior when debugging message formatting or media handling
you can study protocol-related issues instead of guessing
you can build internal tools that mirror real Telegram semantics
you can create reusable infrastructure around updates, not just command handlers

In other words, open source is not just about code availability. It is about having enough visibility to design robust integrations.

Next steps

From here, you can extend the relay in several directions:

add support for callback_query and inline keyboards
implement a /health endpoint and a replay endpoint for admins
send outgoing messages through a separate outbound queue
add per-chat state handling for conversations
migrate the worker to Celery, Dramatiq, or a native message queue

A good next milestone is to support one complete feature end to end:

user sends /start
webhook stores the update
worker handles the command
worker sends a reply
response is logged and traced

That loop is the smallest unit of a serious Telegram integration.

Conclusion

A practical Telegram integration should not stop at “receive a message and print it.” The relay pattern shown here gives you a real foundation: validated webhooks, durable storage, asynchronous processing, and room to scale.

If you are building Telegram tools, bots, or backend automation, this structure will save you time later. It also matches how production systems tend to work: edges are thin, workers are isolated, and update handling is explicit.

Start with this relay, then shape it around your domain logic.

That is where Telegram’s open source ecosystem becomes genuinely useful.