Custom Adapter

Written by
Updated at October 6, 2026

Create a custom adapter when your web framework is not FastAPI or aiohttp. The adapter is the only layer that should know framework-specific request and response types.

FastAPIAdapter and AiohttpAdapter are shipped reference implementations — useful to read, not mandatory templates. Study their source when you need working lifecycle wiring or reply payload handling:

  • aiogram_webhook.web.fastapi — lifespan wiring, Starlette multipart bridge
  • aiogram_webhook.web.aiohttp — web.Application routes and startup/shutdown hooks

See also Extending overview for how adapters fit next to engines and security.

Contract: WebAdapter

Method Responsibility Called when
bind_request(raw) Wrap the framework request as WebRequest. Every webhook POST.
register(app, path, handler, on_startup, on_shutdown) Register POST only; wire lifecycle hooks. engine.register(app).
json_response(status_code, data, headers) Send a JSON body with the framework's own response type: engine errors, empty 200 {} success and a returned TelegramMethod without files. Most responses.
payload_response(status_code, payload, headers) Send an aiohttp Payload (multipart) as the response body. handle_in_background=False with a returned TelegramMethod that uploads files.

WebRequest protocol

Your wrapper must expose:

class MyWebRequest:
    @property
    def raw(self): ...

    @property
    def client_ip(self) -> str | None: ...

    async def json(self) -> dict: ...

    @property
    def headers(self): ...

    @property
    def query_params(self): ...

    @property
    def path_params(self): ...

client_ip feeds IPCheck. path_params must match what your framework extracted for the registered path — the engine does not parse paths itself; Route.match() uses these values.

Minimal skeleton

from aiogram_webhook.web.base import WebAdapter


class MyAdapter(WebAdapter):
    def bind_request(self, request):
        return MyWebRequest(request)

    def register(self, app, path, handler, *, on_startup, on_shutdown) -> None:
        async def endpoint(raw_request):
            return await handler(self.bind_request(raw_request))

        app.post(path, endpoint)
        app.on_startup(on_startup)
        app.on_shutdown(on_shutdown)

    def json_response(self, status_code: int, data=None, headers=None):
        return my_json_response(status_code=status_code, data=data, headers=headers)

    def payload_response(self, status_code: int, payload, headers=None):
        return my_payload_response(status_code=status_code, payload=payload, headers=headers)

Replace app.post / lifecycle hooks with your framework's equivalents. The handler signature is always async (WebRequest) -> FrameworkResponse.

Warning

payload_response() matters when handle_in_background=False. Multipart is required only when the returned TelegramMethod uploads files; it must then be streamed back to Telegram as multipart content, and returning JSON instead will break that foreground reply. Methods without files are sent with json_response().

FastAPI bridges aiohttp Payload through AiohttpPayloadResponse in aiogram_webhook.web._starlette — reuse or adapt that approach on ASGI stacks.

Boundaries: adapter vs engine

Concern Belongs in
Framework route registration Adapter
Request/response type mapping Adapter
Bot resolution Engine
URL construction and matching Route
Secret token and IP verification Security
Telegram setWebhook options WebhookConfig on the engine
Parsing updates and calling aiogram Engine (handle_request)

If verification or bot lookup leaks into the adapter, you duplicate logic that SingleBotEngine / TokenEngine already centralize and tests no longer cover your path.

Framework-specific notes

Stack Practical approach
Starlette (without FastAPI) Mirror FastAPIAdapter: APIRouter or raw routes + lifespan; reuse _starlette.AiohttpPayloadResponse for payloads.
Django (async) Wrap ASGIRequest; register an async view; map JsonResponse and a streaming response for payloads.
Litestar, Quart, etc. Same four-method contract; lifecycle hook names differ per framework.

Register POST only on the webhook path. GET probes from browsers or health checks should not hit the engine handler unless you add them separately.

Using a custom adapter

engine = SingleBotEngine(
    dispatcher,
    bot,
    web=MyAdapter(),
    route=route,
    security=security,
)
engine.register(app)

Engine and Route configuration are identical regardless of the adapter — only web= changes.

Previous
aiohttp
Next
Engines