Telegram Provider¶

Install with: pip install roomkit[telegram] (bundles telegramify-markdown, used to render Markdown into Telegram entities).

Markdown & Rich Messages¶

The provider renders outbound Markdown into native Telegram formatting via telegramify-markdown. By default it sends text with message entities (bold, code, links, etc.); if the converter is unavailable or conversion fails, it falls back to sending the raw text unformatted.

Set rich_messages=True on TelegramConfig to opt in to Bot API 10.1 Rich Messages — native tables and headings — for text sends. Each send tries the Rich Message path first and falls back to entity formatting on any failure (and never re-sends content that already reached the chat):

from roomkit.providers.telegram.config import TelegramConfig

config = TelegramConfig(
    bot_token="...",
    rich_messages=True,   # native tables/headings (Bot API 10.1)
)

Two parsing layers¶

parse_telegram_message(msg) reads a Telegram message object into its parts and attributes nothing:

parts = parse_telegram_message(update["message"])   # TelegramMessageParts | None
parts.content              # TextContent (caption for media) | LocationContent
parts.metadata             # chat_id, date, and the media keys below
parts.message_id
parts.sender_id            # offered, not imposed
parts.entities             # message entities, or caption_entities for media
parts.reply_to_message_id  # what a reply answers, or None
parts.media_group_id       # the album this belongs to, or None

The last three are protocol facts, not attribution: they stay off the InboundMessage, so parse_telegram_webhook's metadata is unchanged.

parse_telegram_webhook is that function plus the ordinary attribution — the sender is message.from.id — wrapped in an InboundMessage. Its external_id and idempotency_key are <chat_id>:<message_id>: Telegram reuses message ids between chats. Malformed nested objects, missing ids and invalid coordinates are rejected with no message rather than escaping an exception from a webhook boundary.

Reach for the lower layer when your identity model is not Telegram's. Under a one-bot-per-user deployment a direct message belongs to the bot's owner, not to the account that typed it, and the same process may apply the opposite rule in a group. Reading a media file's file_id should not cost a consumer its identity model, which is why the layer is public.

Inbound media¶

parse_telegram_webhook handles photo, voice, audio, video_note, video and document. Each parses to a TextContent whose body is the caption — empty when there is none, which is the normal case for a voice note — carrying the file reference in metadata:

Key	Always	Notes
`file_id`	yes	pass to `get_file()`
`media_type`	yes	`voice`, `audio`, `video_note`, `video`, `document`, `photo`
`duration`	no	audio and video kinds
`mime_type`	no	absent on `video_note`
`file_name`	no	`audio`, `video`, `document`
`file_size`	no	bytes

A file_id is only useful to whoever holds the bot token, which is the provider — so the two resolution steps live there:

file_id = inbound.metadata["file_id"]

file_path = await provider.get_file(file_id)        # getFile → path, valid ≥ 1 hour
if file_path:
    data = await provider.download_file(file_path)  # bytes

Both return None on failure and log a warning that never contains the URL — every Bot API URL embeds the bot token. The same rule applies to every API transport error: its safe exception class is returned, never str(exc) with the token-bearing URL. Telegram caps Bot API downloads at 20 MB and refuses larger files at the getFile step, so get_file() returns None for them; metadata["file_size"] lets you tell before spending the call.

RoomKit stops at the bytes. Transcribing a voice note, or storing an attachment, is the application's choice of engine and policy.

The Bot API surface¶

TelegramBotProvider is TelegramBotAPI plus the rendering of a RoomEvent. The API half is what an application needs around its sends — registering a webhook, identifying its own bot, acknowledging a button press, rewriting a message it already sent — so it never writes a second HTTP client for a token the provider already holds.

import os

config = TelegramConfig(
    bot_token="...",
    # Generate this once and persist it in a secret manager.
    webhook_secret=os.environ["TELEGRAM_WEBHOOK_SECRET"],
)
provider = TelegramBotProvider(config)

me = await provider.get_me()
if not me.success:
    ...                              # me.error, me.metadata["description"]
bot = me.metadata["result"]          # id, username, first_name

await provider.set_webhook(
    "https://example.com/hooks/telegram",
    allowed_updates=["message", "callback_query"],       # ask for what you need
)

# In the HTTP endpoint, before parsing or processing the update:
raw = await request.body()
secret_header = request.headers.get("X-Telegram-Bot-Api-Secret-Token", "")
if not provider.verify_signature(raw, secret_header):
    raise HTTPException(status_code=403)

set_webhook() uses TelegramConfig.webhook_secret when secret is omitted. If an explicit secret is supplied to rotate it, the provider starts verifying that value only after Telegram accepts the registration; registration and verification therefore keep one runtime source of truth.

Call	Bot API	Notes
`get_me()`	`getMe`	bot object under `metadata["result"]`
`get_updates(limit, offset)`	`getUpdates`	list under `metadata["result"]`; silent while a webhook is set
`set_webhook(url, secret, allowed_updates, drop_pending_updates)`	`setWebhook`
`delete_webhook(drop_pending_updates)`	`deleteWebhook`
`leave_chat(chat_id)`	`leaveChat`	the only way to stop one chat's updates
`send_message(chat_id, text)`	`sendMessage`	plain text, no Markdown pass
`send_force_reply(chat_id, text)`	`sendMessage`	`provider_message_id` matches the later reply
`send_chat_action(chat_id, action)`	`sendChatAction`	Telegram clears it after ~5s
`answer_callback_query(id, text)`	`answerCallbackQuery`	required, or the button keeps spinning
`edit_message_text(chat_id, message_id, text, reply_markup)`	`editMessageText`	`{"inline_keyboard": []}` drops the buttons
`edit_message_reply_markup(chat_id, message_id, reply_markup)`	`editMessageReplyMarkup`	keyboard only, text untouched

Every one answers with a ProviderResult, so failure reads the same way whichever call produced it: telegram_<code> when Telegram refused, http_<status> when the refusal carried no Bot API body, timeout when nothing came back, or a safe transport exception class — with Telegram's own words under metadata["description"], which is the only text precise enough to say why a webhook URL was rejected. The raw transport exception is never returned because it can contain the token-bearing request URL.

A successful envelope is also checked against the method's result contract: getMe must return a bot object, getUpdates a list of updates with integer ids, boolean mutations the literal true, and sends/edits a Message carrying an integer message_id. A mismatched but otherwise valid JSON payload returns invalid_response instead of leaking a wrong shape to the caller.

Update forms¶

update = parse_telegram_update(payload)     # TelegramUpdate | None
if update is None:
    return                                  # a form you did not ask for
if update.callback:
    press = update.callback                 # TelegramCallback
else:
    parts = parse_telegram_message(update.message)

update.edited tells a new message from an edit of one already delivered — whether that is a new turn or something to ignore is the application's call.

TelegramCallback carries the query id to answer, the data, the sender_id, and the message the button hangs off (chat_id, message_id, message_text) so an outcome can be appended to what was already said. Note that callback_data is posted by whoever pressed the button — any client can send arbitrary bytes to its own bot's webhook — so whatever it names is a claim to check against your own records, never a fact.

Being addressed, and UTF-16¶

if mentions_bot(msg, bot_username="luge_bot", bot_id=42):
    ...

True on any of the five ways Telegram lets someone reach a bot in a busy room: a reply to the bot, an unqualified bot_command (or one qualified with this bot's exact @username), a mention entity, a text_mention naming its id, or the handle posted as plain text with no entity at all. A command qualified for another bot remains false even when Telegram delivers it to this bot, and a longer username does not match by prefix. The helper reports the fact and decides no policy — whether a given group answers only when addressed is your rule.

entity_text(text, entity) slices the stretch an entity covers. Telegram's offsets count UTF-16 code units and Python indexes code points, so text[offset:offset + length] is correct until a character outside the Basic Multilingual Plane appears earlier in the message — one emoji is enough — and silently wrong after.

TelegramBotProvider ¶

TelegramBotProvider(config)

Bases: TelegramBotAPI, TelegramProvider

Send messages via the Telegram Bot API.

verify_signature ¶

verify_signature(payload, signature)

Verify a Telegram webhook secret token.

Telegram's setWebhook accepts a secret_token parameter. On each webhook request Telegram sends the token in the X-Telegram-Bot-Api-Secret-Token header. Verification is a constant-time comparison of that header value against the secret most recently accepted by :meth:set_webhook, initially :attr:TelegramConfig.webhook_secret.

Parameters:

Name	Type	Description	Default
`payload`	`bytes`	Raw request body bytes (unused).	required
`signature`	`str`	Value of the `X-Telegram-Bot-Api-Secret-Token` header.	required

Returns:

Type	Description
`bool`	True if the token matches, False otherwise.

Raises:

Type	Description
`ValueError`	If no webhook secret was configured or registered.

TelegramBotAPI ¶

TelegramBotAPI(config)

Call the Telegram Bot API with a bot token.

Every method answers with a :class:ProviderResult, so a caller reads success and failure the same way whichever call it made: error is telegram_<code> when Telegram refused, http_<status> when the refusal carried no Bot API body, and timeout when nothing came back. Telegram's own words for a refusal — the only text precise enough to tell a caller what to fix — arrive as metadata["description"].

The two reads, :meth:get_me and :meth:get_updates, also carry Telegram's result payload under metadata["result"]. Sends do not: their result is a Message object the caller already has, reduced to the provider_message_id it actually uses.

get_me `async` ¶

get_me()

Identify the bot behind the token — getMe.

The call that tells a good token from a typo. A wrong one comes back telegram_401; an unreachable Telegram as timeout or an http_*. On success metadata["result"] holds the bot object — id, username, first_name.

get_updates `async` ¶

get_updates(*, limit=100, offset=None)

Pull pending updates — getUpdates.

Telegram delivers updates one way or the other, never both: this returns nothing at all while a webhook is registered. Its use is the moment before one is — reading who has already written to the bot.

Parameters:

Name	Type	Description	Default
`limit`	`int`	How many updates to take, at most 100.	`100`
`offset`	`int \| None`	Skip past updates below this id, confirming them as read.	`None`

Returns:

Type	Description
`ProviderResult`	A result whose `metadata["result"]` is the list of raw updates.

set_webhook `async` ¶

set_webhook(url, *, secret=None, allowed_updates=None, drop_pending_updates=False)

Register the URL Telegram POSTs updates to — setWebhook.

Parameters:

Name	Type	Description	Default
`url`	`str`	A public HTTPS URL. Telegram refuses anything else, and refuses a URL it cannot reach — both come back as a failed result whose `metadata["description"]` says which.	required
`secret`	`str \| None`	Echoed back on every request in the `X-Telegram-Bot-Api-Secret-Token` header, which :meth:`TelegramBotProvider.verify_signature` checks. When omitted, :attr:`TelegramConfig.webhook_secret` is used. A successfully registered explicit value becomes the value this provider verifies, so registration and verification cannot silently diverge. Without either value, any host that learns the URL can post updates to it.	`None`
`allowed_updates`	`list[str] \| None`	The update kinds to receive. Telegram's own default omits some kinds entirely, so a consumer that wants `callback_query` must ask for it by name.	`None`
`drop_pending_updates`	`bool`	Discard what queued up while no webhook was registered, rather than delivering it all at once.	`False`

delete_webhook `async` ¶

delete_webhook(*, drop_pending_updates=False)

Stop Telegram POSTing updates — deleteWebhook.

leave_chat `async` ¶

leave_chat(chat_id)

Leave a group, supergroup or channel — leaveChat.

The bot stops receiving that chat's updates. Being removed from a chat is the only way to stop them: a webhook is per-bot, not per-chat.

send_message `async` ¶

send_message(chat_id, text)

Send text as it stands — sendMessage, no Markdown rendering.

A room's outbound traffic goes through :meth:TelegramBotProvider.send, which renders a RoomEvent. This is for text that is already final and belongs to no room — a connection confirmation, an acknowledgement.

send_force_reply `async` ¶

send_force_reply(chat_id, text)

Send a message Telegram opens a reply box under — force_reply.

The answer arrives as an ordinary message carrying reply_to_message.message_id. Matching that against the provider_message_id returned here is what ties an answer back to the question it answers.

send_chat_action `async` ¶

send_chat_action(chat_id, action='typing')

Show a transient status such as "typing…" — sendChatAction.

Telegram clears it after about five seconds, or as soon as the bot sends a real message. Holding it up for a long generation therefore means re-sending it on a shorter cycle than that.

answer_callback_query `async` ¶

answer_callback_query(callback_query_id, text='')

Acknowledge an inline-button press — answerCallbackQuery.

Not optional: until it arrives the client keeps a spinner on the button. text, when given, flashes to the user as a toast.

edit_message_text `async` ¶

edit_message_text(chat_id, message_id, text, *, reply_markup=None)

Rewrite a message already in the chat — editMessageText.

Parameters:

Name	Type	Description	Default
`chat_id`	`str`	The chat holding the message.	required
`message_id`	`int \| str`	The message to rewrite.	required
`text`	`str`	Its new text.	required
`reply_markup`	`dict[str, Any] \| None`	Its new keyboard. Left alone when omitted; `{"inline_keyboard": []}` drops the buttons, which is how a settled prompt stops being clickable.	`None`

edit_message_reply_markup `async` ¶

edit_message_reply_markup(chat_id, message_id, reply_markup)

Replace a message's keyboard and nothing else — editMessageReplyMarkup.

What reflects a toggled selection back to the person who tapped it, without reprinting the message around it.

get_file `async` ¶

get_file(file_id)

Resolve an inbound file_id to a Bot API file path.

The file_id arrives on an inbound update — both :func:~roomkit.providers.telegram.webhook.parse_telegram_message and :func:~roomkit.providers.telegram.webhook.parse_telegram_webhook put it in metadata["file_id"] for every media message. Pair the path this returns with :meth:download_file to get the bytes; the bot token needed for both lives here, not in the calling application.

Telegram keeps the path valid for at least an hour, and refuses any file over 20 MB — the Bot API download ceiling — with a 400. An update carries metadata["file_size"], so a caller can tell which files are past that ceiling without spending the call.

Parameters:

Name	Type	Description	Default
`file_id`	`str`	Identifier from an inbound Telegram update.	required

Returns:

Type	Description
`str \| None`	The file path to download, or None if Telegram refused the file or
`str \| None`	the call failed.

download_file `async` ¶

download_file(file_path)

Download the bytes behind a path returned by :meth:get_file.

Parameters:

Name	Type	Description	Default
`file_path`	`str`	Path returned by :meth:`get_file`.	required

Returns:

Type	Description
`bytes \| None`	The file content, or None if the download failed. Files over the
`bytes \| None`	Bot API's 20 MB ceiling never get this far — :meth:`get_file`
`bytes \| None`	already returned None for them.

TelegramConfig ¶

Bases: BaseModel

Telegram Bot API provider configuration.

connect_timeout `class-attribute` `instance-attribute` ¶

connect_timeout = 5.0

TCP connect timeout in seconds, separate from the request timeout.

file_base_url `property` ¶

file_base_url

Base URL for downloading a file resolved by getFile.

Telegram serves file content from a different path than the one that answers Bot API methods, so this is not a suffix of :attr:base_url. Like it, it embeds the bot token and must never reach a log.

MockTelegramProvider ¶

MockTelegramProvider()

Bases: TelegramProvider

Records sent messages for verification in tests.

parse_telegram_webhook ¶

parse_telegram_webhook(payload, channel_id)

Convert a Telegram Update payload into InboundMessages.

Telegram sends one update at a time (unless using getUpdates). Only message updates are processed; edits, channel posts, and callback queries are silently skipped.

Content is read by :func:parse_telegram_message; what this adds is the attribution — the sender is the Telegram account that typed the message. A consumer whose identity model differs should call that function directly rather than unpick the result here.

Media messages — photo, voice, audio, video_note, video and document — store their file_id in metadata along with a media_type naming the kind, plus whichever of duration, mime_type, file_name and file_size Telegram supplied. The body is the caption, and is empty when there is none. Callers resolve the file_id to bytes via :meth:TelegramBotProvider.get_file and :meth:TelegramBotProvider.download_file.

parse_telegram_message ¶

parse_telegram_message(msg)

Read a Telegram message object into its parts, attributing nothing.

This is the layer to use when your identity model is not Telegram's — it hands back what the message says and leaves who sent it to you. :func:parse_telegram_webhook is this function plus the ordinary attribution.

Text, media (photo, voice, audio, video_note, video, document) and location are understood; anything else — a sticker, a poll — returns None.

Parameters:

Name	Type	Description	Default
`msg`	`dict[str, Any]`	The `message` object of a Telegram Update, not the Update itself.	required

Returns:

Type	Description
`TelegramMessageParts \| None`	The message's parts, or None if it carries no content this understands.

TelegramMessageParts `dataclass` ¶

TelegramMessageParts(content, metadata, message_id, sender_id, entities, reply_to_message_id, media_group_id)

What a Telegram message carries, before anyone decides who sent it.

Attributes:

Name	Type	Description
`content`	`TextContent \| LocationContent`	The message content — text, a caption for media (empty when there is none), or a location.
`metadata`	`dict[str, Any]`	`chat_id` and `date`, plus `file_id`, `media_type` and whichever of `duration`, `mime_type`, `file_name`, `file_size` Telegram supplied for a media message.
`message_id`	`str`	Telegram's message id, as a string.
`sender_id`	`str`	The Telegram account that typed it. Offered, not imposed — a consumer is free to attribute the message to someone else. It is required rather than defaulted so that an unattributed instance is something you write on purpose, not something you forget.
`entities`	`list[dict[str, Any]]`	The message's entities — `caption_entities` for media, since a caption's markup is carried under its own key. Their offsets count UTF-16 code units; :func:`~roomkit.providers.telegram.entities.entity_text` is what slices by them correctly.
`reply_to_message_id`	`str \| None`	The message this one replies to, or None. What ties an answer to the question it answers, when the question was asked with a `force_reply` prompt.
`media_group_id`	`str \| None`	Shared by the several messages Telegram splits one album into, or None. Messages carrying it are one post, delivered as many.

parse_telegram_update ¶

parse_telegram_update(payload)

Read a Telegram Update envelope and say which form it took.

Parameters:

Name	Type	Description	Default
`payload`	`dict[str, Any]`	The Update object, as Telegram POSTs it to a webhook.	required

Returns:

Type	Description
`TelegramUpdate \| None`	The update, or None for a form this does not cover — a channel post, a
`TelegramUpdate \| None`	poll answer, an inline query. Those are kinds a webhook only receives
`TelegramUpdate \| None`	when its `allowed_updates` asked for them.

parse_telegram_callback ¶

parse_telegram_callback(cq)

Read a Telegram callback_query into its parts.

Parameters:

Name	Type	Description	Default
`cq`	`dict[str, Any]`	The `callback_query` object of an Update.	required

Returns:

Type	Description
`TelegramCallback`	The press, with nothing decided about who was allowed to make it.

TelegramUpdate `dataclass` ¶

TelegramUpdate(message, edited, callback)

One Telegram Update, told apart by the form it took.

Exactly one of :attr:message and :attr:callback is set.

Attributes:

Name	Type	Description
`message`	`dict[str, Any] \| None`	The raw `message` object, for a new or edited message.
`edited`	`bool`	True when that message is an edit of one already delivered. The distinction is the application's to act on — some treat an edit as a new turn, others ignore it.
`callback`	`TelegramCallback \| None`	The button press, for a `callback_query` update.

TelegramCallback `dataclass` ¶

TelegramCallback(id, data, sender_id, chat_id, message_id, message_text)

An inline-button press, read but not judged.

Attributes:

Name	Type	Description
`id`	`str`	The callback query's id. Answering it with :meth:`~roomkit.providers.telegram.api.TelegramBotAPI.answer_callback_query` is what stops the spinner on the button.
`data`	`str`	The button's `callback_data`, verbatim. Chosen by whoever built the keyboard, but posted by whoever pressed it — any client can send arbitrary bytes to its own bot's webhook, so what it points at is a claim to check, never a fact.
`sender_id`	`str`	The Telegram account that pressed it.
`chat_id`	`str`	The chat holding the message the button hangs off, empty when Telegram sent no message with the query (an inline-mode result).
`message_id`	`int \| None`	That message's id, or None in the same case.
`message_text`	`str`	That message's text, so an outcome can be appended to what was already said rather than replacing it.

mentions_bot ¶

mentions_bot(msg, *, bot_username=None, bot_id=None)

Say whether a message addresses the bot. The fact, not the policy.

True on any of five things, which are the five ways Telegram lets someone reach a bot in a room full of people:

A reply to a message the bot itself sent.
An unqualified bot_command entity, or one whose @username suffix names this bot. Administrators and bots without privacy mode can receive commands addressed to other bots, so delivery alone is not attribution.
A mention entity whose text is @bot_username.
A text_mention entity naming the bot's user id — how a mention of an account with no username is carried.
@bot_username present as plain text with no entity at all, which is what some clients post.

Whether to answer is not decided here. A group that only responds when addressed asks this; a group that responds to everything never needs to.

Parameters:

Name	Type	Description	Default
`msg`	`dict[str, Any]`	A Telegram `message` object.	required
`bot_username`	`str \| None`	The bot's username, without the `@`. Without it, cases 3 and 5 cannot be checked.	`None`
`bot_id`	`int \| None`	The bot's numeric user id. Without it, cases 1 and 4 cannot be checked.	`None`

Returns:

Type	Description
`bool`	True if the message addresses the bot. False when neither identifier
`bool`	was supplied — nothing can be attributed to a bot that has not said
`bool`	who it is.

entity_text ¶

entity_text(text, entity)

Return the stretch of text an entity covers.

Telegram's offset and length count UTF-16 code units; Python indexes strings by code point. The two agree until a character outside the Basic Multilingual Plane — an emoji, some scripts, a musical symbol — appears earlier in the message, at which point every offset after it is off by one per such character. Encoding to UTF-16-LE puts the slice on the same basis Telegram measured it in.

Parameters:

Name	Type	Description	Default
`text`	`str`	The message text (or caption) the entity indexes into.	required
`entity`	`dict[str, Any]`	A Telegram `MessageEntity`, read for `offset` and `length`.	required

Returns:

Type	Description
`str`	The substring the entity covers, empty when the entity points outside
`str`	the text.

Telegram Provider¶

Markdown & Rich Messages¶

Two parsing layers¶

Inbound media¶

The Bot API surface¶

Update forms¶

Being addressed, and UTF-16¶

TelegramBotProvider ¶

verify_signature ¶

TelegramBotAPI ¶

get_me async ¶

get_updates async ¶

set_webhook async ¶

delete_webhook async ¶

leave_chat async ¶

send_message async ¶

send_force_reply async ¶

send_chat_action async ¶

answer_callback_query async ¶

edit_message_text async ¶

edit_message_reply_markup async ¶

get_file async ¶

download_file async ¶

TelegramConfig ¶

connect_timeout class-attribute instance-attribute ¶

file_base_url property ¶

MockTelegramProvider ¶

parse_telegram_webhook ¶

parse_telegram_message ¶

TelegramMessageParts dataclass ¶

parse_telegram_update ¶

parse_telegram_callback ¶

TelegramUpdate dataclass ¶

TelegramCallback dataclass ¶

mentions_bot ¶

entity_text ¶

get_me `async` ¶

get_updates `async` ¶

set_webhook `async` ¶

delete_webhook `async` ¶

leave_chat `async` ¶

send_message `async` ¶

send_force_reply `async` ¶

send_chat_action `async` ¶

answer_callback_query `async` ¶

edit_message_text `async` ¶

edit_message_reply_markup `async` ¶

get_file `async` ¶

download_file `async` ¶

connect_timeout `class-attribute` `instance-attribute` ¶

file_base_url `property` ¶

TelegramMessageParts `dataclass` ¶

TelegramUpdate `dataclass` ¶

TelegramCallback `dataclass` ¶