Webhook

Telegram Bot API supports webhook. If you set webhook for your bot, Telegram will send updates to the specified url. You can use litegram.methods.set_webhook.SetWebhook() method to specify a url and receive incoming updates on it.

Note

If you use webhook, you can’t use long polling at the same time.

Before start i’ll recommend you to read official Telegram’s documentation about webhook

After you read it, you can start to read this section.

Generally to use webhook with litegram you should use any async web framework. By out of the box litegram has an httpx integration, so we’ll use it.

Note

You can use any async web framework you want, but you should write your own integration if you don’t use httpx.

httpx integration

Out of the box litegram has httpx integration, so you can use it.

Here is available few ways to do it using different implementations of the webhook controller:

You can use it as is or inherit from it and override some methods.

class litegram.webhook.litestar_server.BaseRequestHandler(dispatcher: Dispatcher, handle_in_background: bool = False, **data: Any)[source]
__init__(dispatcher: Dispatcher, handle_in_background: bool = False, **data: Any) → None[source]
class litegram.webhook.litestar_server.SimpleRequestHandler(dispatcher: Dispatcher, bot: Bot, handle_in_background: bool = True, secret_token: str | None = None, **data: Any)[source]
__init__(dispatcher: Dispatcher, bot: Bot, handle_in_background: bool = True, secret_token: str | None = None, **data: Any) → None[source]
class litegram.webhook.litestar_server.TokenBasedRequestHandler(dispatcher: Dispatcher, handle_in_background: bool = True, bot_settings: dict[str, Any] | None = None, **data: Any)[source]
__init__(dispatcher: Dispatcher, handle_in_background: bool = True, bot_settings: dict[str, Any] | None = None, **data: Any) → None[source]

Security

Telegram supports two methods to verify incoming requests that they are from Telegram:

Using a secret token

When you set webhook, you can specify a secret token and then use it to verify incoming requests.

Using IP filtering

You can specify a list of IP addresses from which you expect incoming requests, and then use it to verify incoming requests.

It can be acy using firewall rules or nginx configuration or middleware on application level.

So, litegram has an implementation of the IP filtering for Litestar.

class litegram.webhook.security.IPFilter(ips: Sequence[str | IPv4Network | IPv4Address] | None = None)[source]
__init__(ips: Sequence[str | IPv4Network | IPv4Address] | None = None)[source]

Examples

Behind reverse proxy

In this example we’ll use httpx as web framework and nginx as reverse proxy.

"""
This example shows how to use webhook on behind of any reverse proxy (nginx, traefik, ingress etc.)
"""

from __future__ import annotations

import logging
import sys
from os import getenv
from typing import TYPE_CHECKING, Any

import uvicorn
from litestar import Litestar, Request, post
from litestar.di import Provide

from litegram import Bot, Dispatcher, Router
from litegram.client.default import DefaultBotProperties
from litegram.enums import ParseMode
from litegram.filters import CommandStart
from litegram.types import Update
from litegram.utils.markdown import hbold
from litegram.webhook.litestar_server import webhook_handler

if TYPE_CHECKING:
    from litegram.types import Message

# Bot token can be obtained via https://t.me/BotFather
TOKEN = getenv("BOT_TOKEN")

# Webserver settings
# bind localhost only to prevent any external access
WEB_SERVER_HOST = "127.0.0.1"
# Port for incoming request from reverse proxy. Should be any available port
WEB_SERVER_PORT = 8080

# Path to webhook route, on which Telegram will send requests
WEBHOOK_PATH = "/webhook"
# Secret key to validate requests from Telegram (optional)
WEBHOOK_SECRET = "my-secret"
# Base URL for webhook will be used to generate webhook URL for Telegram,
# in this example it is used public DNS with HTTPS support
BASE_WEBHOOK_URL = "https://litegram.dev"

# All handlers should be attached to the Router (or Dispatcher)
router = Router()


@router.message(CommandStart())
async def command_start_handler(message: Message) -> None:
    """
    This handler receives messages with `/start` command
    """
    # Most event objects have aliases for API methods that can be called in events' context
    # For example if you want to answer to incoming message you can use `message.answer(...)` alias
    # and the target chat will be passed to :ref:`litegram.methods.send_message.SendMessage`
    # method automatically or call API method directly via
    # Bot instance: `bot.send_message(chat_id=message.chat.id, ...)`
    await message.answer(f"Hello, {hbold(message.from_user.full_name)}!")


@router.message()
async def echo_handler(message: Message) -> None:
    """
    Handler will forward receive a message back to the sender

    By default, message handler will handle all message types (like text, photo, sticker etc.)
    """
    try:
        # Send a copy of the received message
        await message.send_copy(chat_id=message.chat.id)
    except TypeError:
        # But not all the types is supported to be copied so need to handle it
        await message.answer("Nice try!")


async def on_startup(bot: Bot) -> None:
    # If you have a self-signed SSL certificate, then you will need to send a public
    # certificate to Telegram
    await bot.set_webhook(f"{BASE_WEBHOOK_URL}{WEBHOOK_PATH}", secret_token=WEBHOOK_SECRET)


def main() -> None:
    # Dispatcher is a root router
    dp = Dispatcher()
    # ... and all other routers should be attached to Dispatcher
    dp.include_router(router)

    # Register startup hook to initialize webhook
    dp.startup.register(on_startup)

    # Initialize Bot instance with default bot properties which will be passed to all API calls
    bot = Bot(token=TOKEN, default=DefaultBotProperties(parse_mode=ParseMode.HTML))

    # And finally start webserver
    app = Litestar(
        route_handlers=[post(WEBHOOK_PATH)(webhook_handler)],
        dependencies={
            "bot": Provide(lambda: bot),
            "dispatcher": Provide(lambda: dp),
            "secret_token": Provide(lambda: WEBHOOK_SECRET),
        },
        on_startup=[lambda: dp.emit_startup(bot=bot)],
        on_shutdown=[lambda: dp.emit_shutdown(bot=bot), bot.session.close],
    )

    uvicorn.run(app, host=WEB_SERVER_HOST, port=WEB_SERVER_PORT)


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO, stream=sys.stdout)
    main()

When you use nginx as reverse proxy, you should set proxy_pass to your httpx server address.

location /webhook {
    proxy_set_header Host $http_host;
    proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
    proxy_redirect off;
    proxy_buffering off;
    proxy_pass http://127.0.0.1:8080;
}

With using other web framework

You can pass incoming request to litegram’s webhook controller from any web framework you want.

Read more about it in litegram.dispatcher.dispatcher.Dispatcher.feed_webhook_update() or litegram.dispatcher.dispatcher.Dispatcher.feed_update() methods.

update = Update.model_validate(await request.json(), context={"bot": bot})
await dispatcher.feed_update(bot, update)

Note

If you want to use reply into webhook, you should check that result of the feed_update methods is an instance of API method and build multipart/form-data or application/json response body manually.