Aiogram Webhooks

aiogram webhooks

aiogram ships one webhook integration: aiohttp, in aiogram.webhook.aiohttp_server. FastAPI, Flask, and others work, but you write the glue.

when to use this skill

• deploying behind HTTPS instead of polling
• TelegramConflictError after switching modes
• securing the webhook endpoint
• serving several bots from one process
• updates arriving but no handler running

polling or webhook

Polling also works in production with a single consumer; choose a webhook when the deployment needs inbound HTTP delivery or multiple replicas. Never both at once on one token — that is what TelegramConflictError means.

minimal webhook app

import os

from aiogram import Bot, Dispatcher, Router
from aiogram.client.default import DefaultBotProperties
from aiogram.enums import ParseMode
from aiogram.filters import CommandStart
from aiogram.types import Message
from aiogram.webhook.aiohttp_server import SimpleRequestHandler, setup_application
from aiohttp import web

WEBHOOK_PATH = "/telegram/webhook"

router = Router(name="webhook-demo")


@router.message(CommandStart())
async def start(message: Message) -> None:
    await message.answer("hello over a webhook")


def build_app(base_url: str, secret: str) -> web.Application:
    bot = Bot(
        token=os.environ["BOT_TOKEN"],
        default=DefaultBotProperties(parse_mode=ParseMode.HTML),
    )
    dispatcher = Dispatcher()
    dispatcher.include_router(router)

    async def on_startup(_: web.Application) -> None:
        await bot.set_webhook(
            url=f"{base_url}{WEBHOOK_PATH}",
            secret_token=secret,
            allowed_updates=dispatcher.resolve_used_update_types(),
            drop_pending_updates=False,
        )

    app = web.Application()
    app.on_startup.append(on_startup)

    SimpleRequestHandler(
        dispatcher=dispatcher,
        bot=bot,
        secret_token=secret,
        handle_in_background=True,
    ).register(app, path=WEBHOOK_PATH)

    setup_application(app, dispatcher, bot=bot)
    return app


application = build_app("https://bot.example.com", "a-long-random-secret")

Run it with web.run_app(application, host="0.0.0.0", port=8080) behind a TLS terminator.

Verified signatures:

SimpleRequestHandler(dispatcher, bot, handle_in_background=True,
                     secret_token=None, **data)
TokenBasedRequestHandler(dispatcher, handle_in_background=True,
                         bot_settings=None, **data)
setup_application(app, dispatcher, /, **kwargs)

setup_application wires the dispatcher's startup and shutdown hooks into aiohttp's lifecycle. Without it, dispatcher lifecycle hooks (including FSM storage cleanup) do not run. SimpleRequestHandler.register() separately registers its own bot-session close hook.

Extra **data on the handler joins workflow data for every update — the webhook equivalent of start_polling(bot, key=value).

security

Telegram will deliver to any URL that answers. Three defences, in order of importance:

1. secret token

import secrets

from aiogram import Bot, Dispatcher
from aiogram.webhook.aiohttp_server import SimpleRequestHandler
from aiohttp import web

secret = secrets.token_urlsafe(32)

app = web.Application()
SimpleRequestHandler(
    dispatcher=Dispatcher(),
    bot=Bot(token="123456:TEST"),
    secret_token=secret,
).register(app, path="/hook")

Telegram echoes the value in X-Telegram-Bot-Api-Secret-Token, and SimpleRequestHandler.verify_secret rejects mismatches with 401. Always set it. Without it, anyone who guesses the URL can inject updates and impersonate any user.

2. an unguessable path

Include a random component in the path. It is defence in depth, not a substitute for the secret token.

3. IP filtering

from aiogram import Dispatcher
from aiogram.webhook.aiohttp_server import ip_filter_middleware
from aiogram.webhook.security import IPFilter
from aiohttp import web

app = web.Application(middlewares=[ip_filter_middleware(IPFilter.default())])
assert isinstance(Dispatcher(), Dispatcher)

IPFilter.default() carries Telegram's published subnets. This middleware reads the leftmost X-Forwarded-For value. Trust it only behind a proxy that replaces untrusted forwarding headers and when the backend cannot be reached directly. Keep the secret header check as well; clients can otherwise forge the forwarded IP.

managing the webhook

from aiogram import Bot


async def switch_to_webhook(bot: Bot, url: str, secret: str) -> None:
    await bot.set_webhook(
        url=url,
        secret_token=secret,
        allowed_updates=["message", "callback_query"],
        max_connections=40,
        drop_pending_updates=False,
    )


async def switch_to_polling(bot: Bot) -> None:
    """Required before polling; otherwise every getUpdates raises Conflict."""
    await bot.delete_webhook(drop_pending_updates=False)


async def diagnose(bot: Bot) -> str:
    info = await bot.get_webhook_info()
    return (
        f"configured={bool(info.url)} pending={info.pending_update_count} "
        f"has_last_error={bool(info.last_error_message)}"
    )

get_webhook_info() is the first thing to check when a webhook bot goes quiet:

drop_pending_updates=True discards the backlog. Convenient in development, destructive in production.

handle_in_background

handle_in_background=True (the default) acknowledges the request immediately and processes the update in a task. After this acknowledgement a crash or handler failure will not cause Telegram to redeliver it. Choose deliberately: important jobs need durable intake before acknowledgement, plus deduplication by bot/update id. Background tasks in aiogram are not a durable queue.

handle_in_background=False processes inline and lets you answer the update with the HTTP response — one fewer API call:

from aiogram import Bot, Dispatcher, Router
from aiogram.methods import SendMessage, TelegramMethod
from aiogram.types import Message

router = Router(name="inline-response")


@router.message()
async def reply_inline(message: Message) -> TelegramMethod[Message]:
    """Returning a method object answers the webhook request directly."""
    return SendMessage(chat_id=message.chat.id, text="pong")


dispatcher = Dispatcher()
dispatcher.include_router(router)
assert isinstance(Bot(token="123456:TEST"), Bot)

Only one method can be returned per update. An inline Bot API response gives the bot no result object or success acknowledgement; make an explicit API call when that result matters. feed_webhook_update also falls back to background processing after its default 55-second timeout, even when the request handler starts inline. Keep this path short.

several bots on one endpoint

from aiogram import Dispatcher
from aiogram.client.default import DefaultBotProperties
from aiogram.webhook.aiohttp_server import TokenBasedRequestHandler, setup_application
from aiohttp import web

dispatcher = Dispatcher()
app = web.Application()

TokenBasedRequestHandler(
    dispatcher=dispatcher,
    bot_settings={"default": DefaultBotProperties(parse_mode="HTML")},
).register(app, path="/webhook/{bot_token}")

setup_application(app, dispatcher)

This is an API construction example, not a secured endpoint. Upstream discourages token-based URLs because proxies can log them. TokenBasedRequestHandler.verify_secret returns True by default; passing secret_token through **data does not enable a check. Prefer separate SimpleRequestHandler endpoints with checked secrets or implement a trusted per-bot resolver and verification before exposing a multi-bot endpoint.

deployment notes

• Cloud Bot API: HTTPS on port 443, 80, 88, or 8443. A local Bot API server has different webhook transport constraints.
• The certificate must be valid, or supplied to set_webhook(certificate=...) if self-signed.
• Set the webhook once, at startup or from a deploy step — not on every replica boot. setWebhook is global for the token.
• With several replicas, all of them serve the same URL behind a load balancer. That is fine; MemoryStorage is not — use Redis, or two replicas will disagree about FSM state.
• Telegram retries failed deliveries. Handlers must be idempotent.

troubleshooting

checklist

• exactly one of polling or webhook per token
• secret_token set and verified
• allowed_updates from resolve_used_update_types()
• setup_application called
• handle_in_background=True unless answering inline deliberately
• persistent FSM storage when running more than one replica
• handlers idempotent against Telegram retries
• delete_webhook() before falling back to polling

see also

• using-aiogram — polling and the run loop
• aiogram-routing — feed_webhook_update and allowed_updates
• aiogram-errors — TelegramConflictError
• aiogram-fsm — why storage matters across replicas