From f4b0e2526198806542a9dcf6e16a2fddbd90f843 Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Fri, 10 Jul 2026 14:20:00 +0200 Subject: [PATCH 1/3] Python: Add Telegram hosting helpers Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 5d6987cd-1b67-4ba1-8b54-3c50da6e7607 --- docs/decisions/0027-hosting-channels.md | 4 +- docs/specs/002-python-hosting-channels.md | 52 ++- python/PACKAGE_STATUS.md | 1 + python/packages/hosting-telegram/LICENSE | 21 ++ python/packages/hosting-telegram/README.md | 83 +++++ .../__init__.py | 43 +++ .../_parsing.py | 338 ++++++++++++++++++ .../_rendering.py | 168 +++++++++ .../packages/hosting-telegram/pyproject.toml | 79 ++++ .../tests/hosting_telegram/test_parsing.py | 269 ++++++++++++++ .../tests/hosting_telegram/test_rendering.py | 226 ++++++++++++ python/pyproject.toml | 1 + .../samples/04-hosting/af-hosting/README.md | 10 +- .../af-hosting/local_telegram/README.md | 143 ++++++++ .../af-hosting/local_telegram/app.py | 303 ++++++++++++++++ .../af-hosting/local_telegram/polling_app.py | 248 +++++++++++++ .../af-hosting/local_telegram/pyproject.toml | 21 ++ python/samples/AGENTS.md | 5 + python/uv.lock | 16 + 19 files changed, 2024 insertions(+), 7 deletions(-) create mode 100644 python/packages/hosting-telegram/LICENSE create mode 100644 python/packages/hosting-telegram/README.md create mode 100644 python/packages/hosting-telegram/agent_framework_hosting_telegram/__init__.py create mode 100644 python/packages/hosting-telegram/agent_framework_hosting_telegram/_parsing.py create mode 100644 python/packages/hosting-telegram/agent_framework_hosting_telegram/_rendering.py create mode 100644 python/packages/hosting-telegram/pyproject.toml create mode 100644 python/packages/hosting-telegram/tests/hosting_telegram/test_parsing.py create mode 100644 python/packages/hosting-telegram/tests/hosting_telegram/test_rendering.py create mode 100644 python/samples/04-hosting/af-hosting/local_telegram/README.md create mode 100644 python/samples/04-hosting/af-hosting/local_telegram/app.py create mode 100644 python/samples/04-hosting/af-hosting/local_telegram/polling_app.py create mode 100644 python/samples/04-hosting/af-hosting/local_telegram/pyproject.toml diff --git a/docs/decisions/0027-hosting-channels.md b/docs/decisions/0027-hosting-channels.md index 55bbc52cfaf..36d4f14b76a 100644 --- a/docs/decisions/0027-hosting-channels.md +++ b/docs/decisions/0027-hosting-channels.md @@ -181,8 +181,8 @@ The app chooses which helper to call for that route and deployment. For example: - `responses_session_id(body)` from `agent-framework-hosting-responses`, which can return either a `resp_*` previous response id or a `conv_*` conversation id when present; -- `telegram_session_id(update)` from `agent-framework-hosting-telegram`, which can choose the chat, user, thread, or - other Telegram-native partitioning logic for that helper; +- `telegram_session_id(update, bot_id=...)` from `agent-framework-hosting-telegram`, which uses the bot and sender for + private chats and the bot and chat for shared group sessions; - `activity_session_id(activity)`, `discord_session_id(interaction_or_message)`, or `a2a_session_id(request_context)` from their respective protocol packages; - `foundry_user_isolation_key()` or `foundry_chat_isolation_key()` from `agent-framework-foundry-hosting`. diff --git a/docs/specs/002-python-hosting-channels.md b/docs/specs/002-python-hosting-channels.md index 190a192fdff..5101982ad8e 100644 --- a/docs/specs/002-python-hosting-channels.md +++ b/docs/specs/002-python-hosting-channels.md @@ -67,7 +67,8 @@ must be aligned with the helper-first model before implementation. Old vocabular |---|---|---| | `agent-framework-hosting` | `agent_framework_hosting` | `AgentState`, `WorkflowState`, `SessionStore`, and run-argument `TypedDict`s. | | `agent-framework-hosting-responses` | `agent_framework_hosting_responses` | Responses helpers: request parsing, session id extraction, response id creation, response rendering, streaming rendering. | -| Future protocol packages | e.g. `agent_framework_hosting_telegram` | Protocol-specific helpers such as `telegram_to_run(...)`, `telegram_from_run(...)`, `telegram_session_id(...)`, and command/media helpers when useful. | +| `agent-framework-hosting-telegram` | `agent_framework_hosting_telegram` | Telegram Bot API helpers: update parsing, chat/session/command/media extraction, final rendering, and streaming edit rendering. | +| Future protocol packages | e.g. `agent_framework_hosting_activity_protocol` | Protocol-specific helpers such as `activity_to_run(...)`, `activity_from_run(...)`, `activity_session_id(...)`, and command/media helpers when useful. | The core hosting package must not depend on protocol SDKs. Protocol packages may depend on their native protocol SDKs if needed, but helper functions should stay usable from plain app code and tests. @@ -244,6 +245,52 @@ text deltas, and a completed event. The final completed payload is produced thro also preserves the model id observed on streaming updates when the finalized `AgentResponse` no longer carries raw model metadata. +## `agent-framework-hosting-telegram` + +The Telegram package provides side-effect-free helpers around Telegram Bot API +update and method payloads. It does not provide a Bot API client, polling loop, +webhook route, command registry, retry policy, or rate limiter. + +### Update helpers + +- `telegram_to_run(update, *, resolve_file_url=None, stream=False) -> AgentRunArgs` +- `telegram_chat_id(update) -> int | None` +- `telegram_session_id(update, *, bot_id) -> str | None` +- `telegram_command(update) -> str | None` +- `telegram_callback_query_id(update) -> str | None` +- `telegram_media_file_id(update_or_message) -> tuple[str, str] | None` + +`telegram_to_run(...)` handles `message`, `edited_message`, and +`callback_query` updates. Text and captions become AF text content. When the +app supplies an async `resolve_file_url` callback, supported Telegram media +file ids can become AF URI content. The package does not call Telegram's +`getFile` method itself. + +`telegram_session_id(..., bot_id=...)` includes the bot identity in every key. +Private chats return `telegram::`; other chats return +`telegram::`, giving groups a shared session by default. This +matches Telegram's native isolation boundaries while preventing two bots from +sharing state accidentally. Apps that want per-user sessions inside a group +can construct a key that includes both chat and sender ids. The app must +authorize those Telegram identities before loading session state. + +`telegram_command(...)` parses Telegram's `/name` and `/name@bot` syntax. It +does not register commands or invoke handlers. + +### Response helpers + +- `telegram_from_run(result, *, chat_id, parse_mode=None)` +- `telegram_from_streaming_run(stream, *, chat_id, message_id, initial_text=None, parse_mode=None)` + +The helpers produce Telegram method/payload values for app-owned Bot API +calls. Final rendering supports text and image URI output and applies +Telegram's text-length boundary. Streaming rendering produces cumulative +`editMessageText` payloads for a placeholder message id supplied by the app, +omitting edits that match an optional `initial_text`, then renders the final +rich output. Image-only responses remove the placeholder with `deleteMessage` +before sending the image. The app owns the initial placeholder send, Bot API +calls, edit throttling, retries, and failure policy. + ## Security responsibilities Protocol helper packages parse and render. They do not authenticate callers, authorize access to state, or decide which @@ -346,3 +393,6 @@ Implementation validation must cover: - Responses streaming SSE rendering; - HTTP round-trip tests showing a native FastAPI route using `AgentState` and Responses helpers; - sample type checking for the local Responses sample. +- Telegram update parsing, chat/session/command/media extraction, final + rendering, and streaming edit rendering; +- sample type checking for the local Telegram polling and webhook entry points. diff --git a/python/PACKAGE_STATUS.md b/python/PACKAGE_STATUS.md index c0080f0cbbf..0cda8f4a1d8 100644 --- a/python/PACKAGE_STATUS.md +++ b/python/PACKAGE_STATUS.md @@ -36,6 +36,7 @@ Status is grouped into these buckets: | `agent-framework-github-copilot` | `python/packages/github_copilot` | `rc` | | `agent-framework-hosting` | `python/packages/hosting` | `alpha` | | `agent-framework-hosting-responses` | `python/packages/hosting-responses` | `alpha` | +| `agent-framework-hosting-telegram` | `python/packages/hosting-telegram` | `alpha` | | `agent-framework-hyperlight` | `python/packages/hyperlight` | `beta` | | `agent-framework-lab` | `python/packages/lab` | `beta` | | `agent-framework-mem0` | `python/packages/mem0` | `beta` | diff --git a/python/packages/hosting-telegram/LICENSE b/python/packages/hosting-telegram/LICENSE new file mode 100644 index 00000000000..9e841e7a26e --- /dev/null +++ b/python/packages/hosting-telegram/LICENSE @@ -0,0 +1,21 @@ + MIT License + + Copyright (c) Microsoft Corporation. + + Permission is hereby granted, free of charge, to any person obtaining a copy + of this software and associated documentation files (the "Software"), to deal + in the Software without restriction, including without limitation the rights + to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + copies of the Software, and to permit persons to whom the Software is + furnished to do so, subject to the following conditions: + + The above copyright notice and this permission notice shall be included in all + copies or substantial portions of the Software. + + THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + SOFTWARE diff --git a/python/packages/hosting-telegram/README.md b/python/packages/hosting-telegram/README.md new file mode 100644 index 00000000000..b267bec0151 --- /dev/null +++ b/python/packages/hosting-telegram/README.md @@ -0,0 +1,83 @@ +# agent-framework-hosting-telegram + +Telegram Bot API-shaped helpers for app-owned Agent Framework hosting. + +This is an alpha, helper-only package: it converts between Telegram's +`Update` JSON shape and Agent Framework run values in both directions. It +does **not** provide a bot client, a hosting/channel registry, or a +long-running service. Your app remains fully responsible for: + +- **Fetching updates** -- long polling (`getUpdates`) or registering a + webhook -- and for verifying webhook authenticity (e.g. Telegram's secret + token header, or an IP allowlist). +- **The Bot API client** -- issuing the actual HTTP calls (`sendMessage`, + `sendPhoto`, `editMessageText`, `answerCallbackQuery`, `getFile`, ...) with + whatever HTTP library you prefer. +- **Rate limits and retries** -- Telegram enforces per-chat and global rate + limits; back off and retry on `429`/`5xx` yourself. +- **Command dispatch** -- `telegram_command(...)` only parses and normalizes + a leading `/command`; your app decides what each command does. +- **Sessions/storage** -- pair these helpers with + [`agent-framework-hosting`](https://pypi.org/project/agent-framework-hosting/)'s + `AgentState` / `SessionStore` (or your own) to persist `AgentSession`s across turns. + +## Helpers + +- `telegram_chat_id(update)` -- the chat id an update belongs to. +- `telegram_session_id(update, bot_id=...)` -- a bot-scoped `AgentState` + session id. Private chats use `telegram::`; other chats use + `telegram::`. +- `telegram_command(update)` -- a leading slash command, with `/name@bot args` + normalized to `/name args`. Returns `None` if there is none. +- `telegram_callback_query_id(update)` -- a callback query's id, so you can + call `answerCallbackQuery` yourself. +- `telegram_media_file_id(update_or_message)` -- the `(file_id, mime_type)` + of inbound media (largest photo size, document, voice, audio, or video). +- `telegram_to_run(update, *, resolve_file_url=None, stream=False)` -- convert + a `message`, `edited_message`, or `callback_query` update into + `Agent.run` arguments. Provide `resolve_file_url` (typically backed by + `getFile`) to turn inbound media into content; without it (or when it + returns `None`), text/caption is preserved and media is otherwise dropped. + Media-only input with no resolvable URL raises `ValueError`. +- `telegram_from_run(result, *, chat_id, parse_mode=None)` -- render a + finished run as one `TelegramOperation` (`sendPhoto` when the response has + an image, otherwise `sendMessage`, falling back to `"(no response)"`). +- `telegram_from_streaming_run(stream, *, chat_id, message_id, + initial_text=None, parse_mode=None)` -- render a streaming run as + `editMessageText` operations with the cumulative text so far, followed by any + images in the final response as `sendPhoto` operations. Pass the + app-created placeholder text as `initial_text` so an identical first edit is + omitted. Image-only responses first emit `deleteMessage` for the placeholder. + +`TelegramOperation` is a minimal `TypedDict` of `{"method": str, "payload": dict}` +-- your app is responsible for actually calling the Bot API with it. + +```python +from agent_framework_hosting import AgentState +from agent_framework_hosting_telegram import ( + telegram_chat_id, + telegram_from_run, + telegram_session_id, + telegram_to_run, +) + +state = AgentState(agent) + + +async def handle_update(update: dict) -> None: + chat_id = telegram_chat_id(update) + if chat_id is None: + return # Not a chat update this bot handles. + + session_id = telegram_session_id(update, bot_id=bot.id) + session = await state.get_or_create_session(session_id) + run = await telegram_to_run(update, resolve_file_url=resolve_telegram_file_url) + result = await (await state.get_target()).run(run["messages"], session=session, options=run["options"]) + await state.set_session(session_id, session) # type: ignore[arg-type] + + operation = telegram_from_run(result, chat_id=chat_id) + await call_bot_api(operation["method"], operation["payload"]) # Your HTTP client. +``` + +The base execution-state helpers live in +[`agent-framework-hosting`](https://pypi.org/project/agent-framework-hosting/). diff --git a/python/packages/hosting-telegram/agent_framework_hosting_telegram/__init__.py b/python/packages/hosting-telegram/agent_framework_hosting_telegram/__init__.py new file mode 100644 index 00000000000..90ac4c8f654 --- /dev/null +++ b/python/packages/hosting-telegram/agent_framework_hosting_telegram/__init__.py @@ -0,0 +1,43 @@ +# Copyright (c) Microsoft. All rights reserved. + +"""Telegram Bot API-shaped helpers for app-owned Agent Framework hosting.""" + +import importlib.metadata + +from ._parsing import ( + ResolveFileUrl, + telegram_callback_query_id, + telegram_chat_id, + telegram_command, + telegram_media_file_id, + telegram_session_id, + telegram_to_run, +) +from ._rendering import ( + TELEGRAM_MAX_CAPTION_LENGTH, + TELEGRAM_MAX_TEXT_LENGTH, + TelegramOperation, + telegram_from_run, + telegram_from_streaming_run, +) + +try: + __version__ = importlib.metadata.version(__name__) +except importlib.metadata.PackageNotFoundError: + __version__ = "0.0.0" + +__all__ = [ + "TELEGRAM_MAX_CAPTION_LENGTH", + "TELEGRAM_MAX_TEXT_LENGTH", + "ResolveFileUrl", + "TelegramOperation", + "__version__", + "telegram_callback_query_id", + "telegram_chat_id", + "telegram_command", + "telegram_from_run", + "telegram_from_streaming_run", + "telegram_media_file_id", + "telegram_session_id", + "telegram_to_run", +] diff --git a/python/packages/hosting-telegram/agent_framework_hosting_telegram/_parsing.py b/python/packages/hosting-telegram/agent_framework_hosting_telegram/_parsing.py new file mode 100644 index 00000000000..361d8418676 --- /dev/null +++ b/python/packages/hosting-telegram/agent_framework_hosting_telegram/_parsing.py @@ -0,0 +1,338 @@ +# Copyright (c) Microsoft. All rights reserved. + +"""Parsing helpers for the Telegram Bot API ``Update`` object. + +Telegram delivers updates as JSON objects shaped like ``{"update_id": ..., +"message": {...}}`` (or ``edited_message`` / ``callback_query`` instead of +``message``). These helpers pull out the handful of fields an app-owned route +needs -- chat id, a namespaced session id, a leading slash command, a +callback query id, and inbound media -- and translate an update into Agent +Framework ``Agent.run`` arguments. They do not poll for updates, register +webhooks, call the Telegram HTTP API, or dispatch commands; that is app-owned +route/bot-client code. +""" + +from __future__ import annotations + +import re +from collections.abc import Awaitable, Callable, Mapping, Sequence +from typing import Any, cast + +from agent_framework import ChatOptions, Content, Message +from agent_framework_hosting import AgentRunArgs + +# Telegram media fields whose objects carry a `file_id` (and, except photos, +# a `mime_type`) directly, mapped to the MIME type Telegram uses when the +# object omits `mime_type` (voice notes are always OGG/Opus, for example). +_MEDIA_DEFAULT_MIME_TYPES: dict[str, str] = { + "document": "application/octet-stream", + "voice": "audio/ogg", + "audio": "audio/mpeg", + "video": "video/mp4", +} + +# Matches a leading Telegram bot command: `/name`, optionally `@botname`, +# optionally followed by arguments. Command names are `[A-Za-z0-9_]` per the +# Bot API's `bot_command` entity rules. +_COMMAND_PATTERN = re.compile(r"^/(?P[A-Za-z0-9_]+)(?:@(?P[A-Za-z0-9_]+))?(?P.*)$", re.DOTALL) + +ResolveFileUrl = Callable[[str], Awaitable[str | None]] +"""Async callable resolving a Telegram ``file_id`` to a fetchable URL, or ``None``.""" + + +def _inner_message(update: Mapping[str, Any]) -> Mapping[str, Any] | None: + """Return the ``message`` or ``edited_message`` object from ``update``, if present.""" + for key in ("message", "edited_message"): + candidate = update.get(key) + if isinstance(candidate, Mapping): + return cast("Mapping[str, Any]", candidate) + return None + + +def _callback_query(update: Mapping[str, Any]) -> Mapping[str, Any] | None: + """Return the ``callback_query`` object from ``update``, if present.""" + candidate = update.get("callback_query") + return cast("Mapping[str, Any]", candidate) if isinstance(candidate, Mapping) else None + + +def telegram_chat_id(update: Mapping[str, Any]) -> int | None: + """Return the chat id an update belongs to. + + Reads ``message.chat.id`` / ``edited_message.chat.id``, falling back to + ``callback_query.message.chat.id`` for callback-only updates. + + Args: + update: A Telegram Bot API ``Update`` object. + + Returns: + The chat id, or ``None`` if the update carries none of the supported shapes. + """ + message = _inner_message(update) + if message is None: + callback_query = _callback_query(update) + candidate = callback_query.get("message") if callback_query is not None else None + if not isinstance(candidate, Mapping): + return None + message = cast("Mapping[str, Any]", candidate) + chat_candidate = message.get("chat") + if not isinstance(chat_candidate, Mapping): + return None + chat = cast("Mapping[str, Any]", chat_candidate) + chat_id = chat.get("id") + return chat_id if isinstance(chat_id, int) else None + + +def telegram_session_id(update: Mapping[str, Any], *, bot_id: int) -> str | None: + """Return a session id using Telegram's native bot, user, and chat boundaries. + + Args: + update: A Telegram Bot API ``Update`` object. + + Keyword Args: + bot_id: The Telegram bot's numeric user id. + + Returns: + ``telegram::`` for a private chat, + ``telegram::`` for other chats, or ``None`` when the + required Telegram identity is absent. + """ + chat_id = telegram_chat_id(update) + if chat_id is None: + return None + + message = _inner_message(update) + callback_query = _callback_query(update) + if message is None and callback_query is not None: + callback_message = callback_query.get("message") + if isinstance(callback_message, Mapping): + message = cast("Mapping[str, Any]", callback_message) + + chat_candidate = message.get("chat") if message is not None else None + chat = cast("Mapping[str, Any]", chat_candidate) if isinstance(chat_candidate, Mapping) else None + chat_type = chat.get("type") if chat is not None else None + if chat_type != "private": + return f"telegram:{bot_id}:{chat_id}" + + if callback_query is not None: + sender_candidate = callback_query.get("from") + elif message is not None: + sender_candidate = message.get("from") + else: + return None + if not isinstance(sender_candidate, Mapping): + return None + sender = cast("Mapping[str, Any]", sender_candidate) + sender_id = sender.get("id") + return f"telegram:{bot_id}:{sender_id}" if isinstance(sender_id, int) else None + + +def telegram_callback_query_id(update: Mapping[str, Any]) -> str | None: + """Return the callback query id from an update, if present. + + Apps need this to answer the callback query (``answerCallbackQuery``) so + Telegram stops showing the client-side loading spinner; this helper only + extracts the id, it does not call the Bot API. + + Args: + update: A Telegram Bot API ``Update`` object. + + Returns: + The callback query id, or ``None`` if the update has no ``callback_query``. + """ + callback_query = _callback_query(update) + if callback_query is None: + return None + query_id = callback_query.get("id") + return query_id if isinstance(query_id, str) else None + + +def _command_source_text(update: Mapping[str, Any]) -> str | None: + """Return the text a leading command should be parsed from.""" + message = _inner_message(update) + if message is not None: + text = message.get("text") + if isinstance(text, str): + return text + callback_query = _callback_query(update) + if callback_query is not None: + data = callback_query.get("data") + if isinstance(data, str): + return data + return None + + +def telegram_command(update: Mapping[str, Any]) -> str | None: + """Parse a leading slash command out of an update, without dispatching it. + + Looks at ``message.text`` / ``edited_message.text`` first, then + ``callback_query.data``. A bot-suffixed command (``/name@bot args``) is + normalized to ``/name args`` since a single Bot API integration only ever + serves one bot username. Callers are responsible for matching the + returned command name and acting on it. + + Args: + update: A Telegram Bot API ``Update`` object. + + Returns: + The normalized command (e.g. ``"/start"`` or ``"/start hello"``), or + ``None`` if the source text does not start with a command. + """ + text = _command_source_text(update) + if text is None: + return None + match = _COMMAND_PATTERN.match(text) + if not match: + return None + return f"/{match.group('name')}{match.group('rest')}".rstrip() + + +def _resolve_media_message(update_or_message: Mapping[str, Any]) -> Mapping[str, Any] | None: + """Return the message object to inspect for media, from either an update or a bare message.""" + if "message" in update_or_message or "edited_message" in update_or_message or "callback_query" in update_or_message: + message = _inner_message(update_or_message) + if message is not None: + return message + callback_query = _callback_query(update_or_message) + if callback_query is not None: + candidate = callback_query.get("message") + if isinstance(candidate, Mapping): + return cast("Mapping[str, Any]", candidate) + return None + return update_or_message + + +def _largest_photo_file_id(photo_sizes: Sequence[Any]) -> str | None: + """Return the ``file_id`` of the highest-resolution entry in a Telegram ``photo`` array.""" + candidates = [cast("Mapping[str, Any]", size) for size in photo_sizes if isinstance(size, Mapping)] + if not candidates: + return None + + def _area(size: Mapping[str, Any]) -> int: + file_size = size.get("file_size") + if isinstance(file_size, int): + return file_size + width, height = size.get("width"), size.get("height") + return width * height if isinstance(width, int) and isinstance(height, int) else 0 + + largest = max(candidates, key=_area) + file_id = largest.get("file_id") + return file_id if isinstance(file_id, str) else None + + +def telegram_media_file_id(update_or_message: Mapping[str, Any]) -> tuple[str, str] | None: + """Return the ``(file_id, mime_type)`` of the inbound media attached to an update or message. + + Accepts either a full ``Update`` object (media is read from ``message`` / + ``edited_message`` / ``callback_query.message``) or a bare Telegram + message object. Photos pick the largest size Telegram sent (by + ``file_size``, falling back to pixel area); documents, voice notes, + audio, and video use their own ``file_id`` and ``mime_type`` (falling + back to Telegram's known default MIME type when the field is absent). + + Args: + update_or_message: A Telegram ``Update`` or message object. + + Returns: + A ``(file_id, mime_type)`` tuple, or ``None`` if there is no supported media. + """ + message = _resolve_media_message(update_or_message) + if message is None: + return None + + photo = message.get("photo") + if isinstance(photo, Sequence) and not isinstance(photo, (str, bytes, bytearray)) and photo: + file_id = _largest_photo_file_id(cast("Sequence[Any]", photo)) + if file_id is not None: + return file_id, "image/jpeg" + + for key, default_mime_type in _MEDIA_DEFAULT_MIME_TYPES.items(): + item_candidate = message.get(key) + if isinstance(item_candidate, Mapping): + item = cast("Mapping[str, Any]", item_candidate) + file_id = item.get("file_id") + if isinstance(file_id, str): + mime_type = item.get("mime_type") + return file_id, mime_type if isinstance(mime_type, str) and mime_type else default_mime_type + return None + + +async def _contents_from_message( + message: Mapping[str, Any], + resolve_file_url: ResolveFileUrl | None, +) -> list[Content]: + """Translate one Telegram message object into Agent Framework content parts. + + Raises: + ValueError: If the message has no text, caption, or resolvable media. + """ + text = message.get("text") + caption = message.get("caption") + text_value = text if isinstance(text, str) and text else (caption if isinstance(caption, str) and caption else None) + + contents: list[Content] = [] + media = telegram_media_file_id(message) + if media is not None: + file_id, mime_type = media + resolved_url = await resolve_file_url(file_id) if resolve_file_url is not None else None + if resolved_url: + contents.append(Content.from_uri(uri=resolved_url, media_type=mime_type)) + elif text_value is None: + raise ValueError( + f"Cannot resolve Telegram media file_id={file_id!r}: no `resolve_file_url` was given (or it " + "returned None), and the message has no text/caption to fall back to." + ) + if text_value is not None: + contents.append(Content.from_text(text=text_value)) + + if not contents: + raise ValueError("Telegram message has no text, caption, or resolvable media to convert to a run.") + return contents + + +async def telegram_to_run( + update: Mapping[str, Any], + *, + resolve_file_url: ResolveFileUrl | None = None, + stream: bool = False, +) -> AgentRunArgs: + """Convert a Telegram update into Agent Framework run values. + + Supports ``message``, ``edited_message``, and ``callback_query`` updates. + Message/edited-message text or caption becomes text content; inbound + media becomes uri content when ``resolve_file_url`` resolves it (media + alone with no resolvable URL and no text/caption raises rather than + producing an empty run). A callback query's ``data`` becomes the user's + text. + + Args: + update: A Telegram Bot API ``Update`` object. + + Keyword Args: + resolve_file_url: Optional async callable that resolves a Telegram + ``file_id`` (typically via the Bot API's ``getFile``) to a + fetchable URL, or ``None`` if it cannot. When omitted, media is + ignored and only text/caption is used. + stream: Whether the caller intends to run the agent in streaming mode. + + Returns: + Arguments corresponding to ``Agent.run``. + + Raises: + ValueError: If the update has no actionable message/callback data, or + a message has no text, caption, or resolvable media. + """ + message = _inner_message(update) + if message is not None: + contents = await _contents_from_message(message, resolve_file_url) + else: + callback_query = _callback_query(update) + data = callback_query.get("data") if callback_query is not None else None + if not isinstance(data, str) or not data: + raise ValueError("Telegram update has no actionable `message`, `edited_message`, or `callback_query.data`.") + contents = [Content.from_text(text=data)] + + return AgentRunArgs( + messages=[Message("user", contents)], + options=cast("ChatOptions[Any]", {}), + stream=stream, + ) diff --git a/python/packages/hosting-telegram/agent_framework_hosting_telegram/_rendering.py b/python/packages/hosting-telegram/agent_framework_hosting_telegram/_rendering.py new file mode 100644 index 00000000000..711225ac868 --- /dev/null +++ b/python/packages/hosting-telegram/agent_framework_hosting_telegram/_rendering.py @@ -0,0 +1,168 @@ +# Copyright (c) Microsoft. All rights reserved. + +"""Rendering helpers that translate Agent Framework results into native Telegram Bot API calls. + +Each helper returns (or yields) a small ``TelegramOperation`` -- a Telegram +Bot API method name plus its JSON payload. App-owned code decides how to +actually invoke the Bot API (``sendMessage``, ``sendPhoto``, +``editMessageText``, ...), including authentication, retries, and rate +limiting; these helpers make no network calls. +""" + +from __future__ import annotations + +from collections.abc import AsyncIterator +from typing import Any, TypedDict + +from agent_framework import AgentResponse, AgentResponseUpdate, ResponseStream + +# Telegram's documented maximum length, in UTF-16 code units, for message +# text (`sendMessage` / `editMessageText`) and photo captions (`sendPhoto`). +# Agent Framework text is plain `str`; truncating by Python character count +# is a conservative (never-too-long) approximation of the UTF-16 limit. +TELEGRAM_MAX_TEXT_LENGTH = 4096 +TELEGRAM_MAX_CAPTION_LENGTH = 1024 + +_NO_RESPONSE_TEXT = "(no response)" + + +class TelegramOperation(TypedDict): + """A single native Telegram Bot API call produced by a rendering helper. + + Attributes: + method: The Telegram Bot API method name (e.g. ``"sendMessage"``). + payload: The JSON-serializable request body for that method. + """ + + method: str + payload: dict[str, Any] + + +def _truncate(text: str, max_length: int) -> str: + """Deterministically cap ``text`` at ``max_length`` characters.""" + return text if len(text) <= max_length else text[:max_length] + + +def _text_and_image_uris(result: AgentResponse[Any]) -> tuple[str, list[str]]: + """Return the response's concatenated text and any image uris it carries.""" + image_uris = [ + content.uri + for message in result.messages + for content in message.contents + if content.type in ("data", "uri") and content.uri and (content.media_type or "").startswith("image/") + ] + return result.text, image_uris + + +def telegram_from_run( + result: AgentResponse[Any], + *, + chat_id: int, + parse_mode: str | None = None, +) -> TelegramOperation: + """Render a finished agent run as one native Telegram Bot API call. + + An image in the response renders as ``sendPhoto`` (using the first image + found; any accompanying text becomes its caption). Otherwise, response + text renders as ``sendMessage``, falling back to ``"(no response)"`` when + the response has neither text nor an image. + + Args: + result: The finished agent response to render. + + Keyword Args: + chat_id: The Telegram chat id to address the message to. + parse_mode: Optional Telegram ``parse_mode`` (e.g. ``"MarkdownV2"``, + ``"HTML"``) to attach to the rendered payload. + + Returns: + A ``TelegramOperation`` describing the Bot API call to make. + """ + text, image_uris = _text_and_image_uris(result) + if image_uris: + payload: dict[str, Any] = {"chat_id": chat_id, "photo": image_uris[0]} + if text: + payload["caption"] = _truncate(text, TELEGRAM_MAX_CAPTION_LENGTH) + if parse_mode: + payload["parse_mode"] = parse_mode + return TelegramOperation(method="sendPhoto", payload=payload) + + payload = {"chat_id": chat_id, "text": _truncate(text or _NO_RESPONSE_TEXT, TELEGRAM_MAX_TEXT_LENGTH)} + if parse_mode: + payload["parse_mode"] = parse_mode + return TelegramOperation(method="sendMessage", payload=payload) + + +async def telegram_from_streaming_run( + stream: ResponseStream[AgentResponseUpdate, AgentResponse[Any]], + *, + chat_id: int, + message_id: int, + initial_text: str | None = None, + parse_mode: str | None = None, +) -> AsyncIterator[TelegramOperation]: + """Render a streaming agent run as a sequence of native Telegram Bot API calls. + + Yields one ``editMessageText`` operation per update carrying new text, each + with the cumulative text so far (Telegram has no incremental-append edit + call). Interim edits omit ``parse_mode`` since intermediate text is not + guaranteed to be valid in the target markup. After the stream ends, yields + a final ``editMessageText`` reflecting the finalized response text (this + one includes ``parse_mode`` when given) and then one ``sendPhoto`` + operation per image the finalized response carries. Errors raised while + iterating the stream or while finalizing it (``stream.get_final_response()``) + propagate to the caller rather than being swallowed. + + Args: + stream: The agent response stream returned by ``agent.run(..., stream=True)``. + + Keyword Args: + chat_id: The Telegram chat id the streamed message lives in. + message_id: The id of the Telegram message to edit as new content arrives. + initial_text: The current text of the app-created placeholder message. + Matching edits are omitted because Telegram rejects no-op edits. + parse_mode: Optional Telegram ``parse_mode`` to attach to the final text edit. + + Yields: + ``TelegramOperation`` values describing the Bot API calls to make, in order. + """ + text = "" + last_rendered_text = _truncate(initial_text, TELEGRAM_MAX_TEXT_LENGTH) if initial_text is not None else "" + async for update in stream: + if update.text: + text += update.text + rendered_text = _truncate(text, TELEGRAM_MAX_TEXT_LENGTH) + if rendered_text == last_rendered_text: + continue + last_rendered_text = rendered_text + yield TelegramOperation( + method="editMessageText", + payload={ + "chat_id": chat_id, + "message_id": message_id, + "text": rendered_text, + }, + ) + + final = await stream.get_final_response() + final_text, image_uris = _text_and_image_uris(final) + + if final_text or not image_uris: + rendered_text = _truncate(final_text or _NO_RESPONSE_TEXT, TELEGRAM_MAX_TEXT_LENGTH) + payload: dict[str, Any] = { + "chat_id": chat_id, + "message_id": message_id, + "text": rendered_text, + } + if parse_mode: + payload["parse_mode"] = parse_mode + if rendered_text != last_rendered_text or parse_mode: + yield TelegramOperation(method="editMessageText", payload=payload) + else: + yield TelegramOperation( + method="deleteMessage", + payload={"chat_id": chat_id, "message_id": message_id}, + ) + + for uri in image_uris: + yield TelegramOperation(method="sendPhoto", payload={"chat_id": chat_id, "photo": uri}) diff --git a/python/packages/hosting-telegram/pyproject.toml b/python/packages/hosting-telegram/pyproject.toml new file mode 100644 index 00000000000..3c0584da3cd --- /dev/null +++ b/python/packages/hosting-telegram/pyproject.toml @@ -0,0 +1,79 @@ +[project] +name = "agent-framework-hosting-telegram" +description = "Telegram Bot API-shaped helpers for agent-framework-hosting." +authors = [{ name = "Microsoft", email = "af-support@microsoft.com"}] +readme = "README.md" +requires-python = ">=3.10" +version = "1.0.0a260709" +license-files = ["LICENSE"] +urls.homepage = "https://aka.ms/agent-framework" +urls.source = "https://github.com/microsoft/agent-framework/tree/main/python" +urls.release_notes = "https://github.com/microsoft/agent-framework/releases?q=tag%3Apython-1&expanded=true" +urls.issues = "https://github.com/microsoft/agent-framework/issues" +classifiers = [ + "License :: OSI Approved :: MIT License", + "Development Status :: 3 - Alpha", + "Intended Audience :: Developers", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Programming Language :: Python :: 3.13", + "Programming Language :: Python :: 3.14", + "Typing :: Typed", +] +dependencies = [ + "agent-framework-core>=1.11.0,<2", + "agent-framework-hosting==1.0.0a260709", +] + +[tool.uv] +prerelease = "if-necessary-or-explicit" +environments = [ + "sys_platform == 'darwin'", + "sys_platform == 'linux'", + "sys_platform == 'win32'" +] + +[tool.uv-dynamic-versioning] +fallback-version = "0.0.0" + +[tool.pytest.ini_options] +testpaths = 'tests' +addopts = "-ra -q -r fEX" +asyncio_mode = "auto" +asyncio_default_fixture_loop_scope = "function" +filterwarnings = [] +timeout = 120 +markers = [ + "integration: marks tests as integration tests that require external services", +] + +[tool.ruff] +extend = "../../pyproject.toml" + +[tool.coverage.run] +omit = [ + "**/__init__.py" +] + +[tool.pyright] +extends = "../../pyproject.toml" +include = ["agent_framework_hosting_telegram"] +exclude = ['tests'] + +[tool.bandit] +targets = ["agent_framework_hosting_telegram"] +exclude_dirs = ["tests"] + +[tool.poe] +executor.type = "uv" +include = "../../shared_tasks.toml" + +[tool.poe.tasks.test] +help = "Run the default unit test suite for this package." +cmd = 'pytest -m "not integration" --cov=agent_framework_hosting_telegram --cov-report=term-missing:skip-covered tests' + +[build-system] +requires = ["flit-core >= 3.11,<4.0"] +build-backend = "flit_core.buildapi" diff --git a/python/packages/hosting-telegram/tests/hosting_telegram/test_parsing.py b/python/packages/hosting-telegram/tests/hosting_telegram/test_parsing.py new file mode 100644 index 00000000000..0cb4af0041b --- /dev/null +++ b/python/packages/hosting-telegram/tests/hosting_telegram/test_parsing.py @@ -0,0 +1,269 @@ +# Copyright (c) Microsoft. All rights reserved. + +"""Tests for the Telegram update parsing/extraction helpers.""" + +from __future__ import annotations + +from typing import Any, cast + +import pytest +from agent_framework import Message + +from agent_framework_hosting_telegram import ( + telegram_callback_query_id, + telegram_chat_id, + telegram_command, + telegram_media_file_id, + telegram_session_id, + telegram_to_run, +) + + +def _message_update(**message_fields: Any) -> dict[str, Any]: + return { + "update_id": 1, + "message": { + "message_id": 10, + "date": 0, + "chat": {"id": 555, "type": "private"}, + "from": {"id": 777, "is_bot": False, "first_name": "Ada"}, + **message_fields, + }, + } + + +class TestTelegramChatId: + def test_reads_from_message(self) -> None: + assert telegram_chat_id(_message_update(text="hi")) == 555 + + def test_reads_from_edited_message(self) -> None: + update = {"update_id": 1, "edited_message": {"chat": {"id": 42}, "text": "hi"}} + assert telegram_chat_id(update) == 42 + + def test_reads_from_callback_query_message(self) -> None: + update = { + "update_id": 1, + "callback_query": {"id": "cb1", "data": "x", "message": {"chat": {"id": 99}}}, + } + assert telegram_chat_id(update) == 99 + + def test_returns_none_when_absent(self) -> None: + assert telegram_chat_id({"update_id": 1}) is None + + def test_returns_none_for_non_int_chat_id(self) -> None: + update = {"update_id": 1, "message": {"chat": {"id": "not-an-int"}}} + assert telegram_chat_id(update) is None + + +class TestTelegramSessionId: + def test_private_chat_uses_bot_and_user_id(self) -> None: + assert telegram_session_id(_message_update(text="hi"), bot_id=123) == "telegram:123:777" + + def test_returns_none_when_no_chat_id(self) -> None: + assert telegram_session_id({"update_id": 1}, bot_id=123) is None + + def test_private_chat_returns_none_when_no_user_id(self) -> None: + update = _message_update(text="hi") + del update["message"]["from"] + assert telegram_session_id(update, bot_id=123) is None + + def test_group_chat_uses_bot_and_chat_id(self) -> None: + update = _message_update(text="hi") + update["message"]["chat"] = {"id": -555, "type": "supergroup"} + assert telegram_session_id(update, bot_id=123) == "telegram:123:-555" + + def test_private_callback_uses_callback_sender(self) -> None: + update = { + "update_id": 1, + "callback_query": { + "id": "cb1", + "from": {"id": 888}, + "message": {"chat": {"id": 555, "type": "private"}}, + }, + } + assert telegram_session_id(update, bot_id=123) == "telegram:123:888" + + +class TestTelegramCommand: + def test_plain_command(self) -> None: + assert telegram_command(_message_update(text="/start")) == "/start" + + def test_command_with_args(self) -> None: + assert telegram_command(_message_update(text="/echo hello world")) == "/echo hello world" + + def test_bot_suffixed_command_normalizes(self) -> None: + assert telegram_command(_message_update(text="/start@mybot")) == "/start" + + def test_bot_suffixed_command_with_args_normalizes(self) -> None: + assert telegram_command(_message_update(text="/echo@mybot hello")) == "/echo hello" + + def test_edited_message_text(self) -> None: + update = {"update_id": 1, "edited_message": {"chat": {"id": 1}, "text": "/help"}} + assert telegram_command(update) == "/help" + + def test_callback_query_data(self) -> None: + update = {"update_id": 1, "callback_query": {"id": "cb1", "data": "/confirm@mybot yes"}} + assert telegram_command(update) == "/confirm yes" + + def test_non_command_text_returns_none(self) -> None: + assert telegram_command(_message_update(text="hello there")) is None + + def test_missing_text_returns_none(self) -> None: + assert telegram_command({"update_id": 1, "message": {"chat": {"id": 1}}}) is None + + def test_no_actionable_source_returns_none(self) -> None: + assert telegram_command({"update_id": 1}) is None + + +class TestTelegramCallbackQueryId: + def test_returns_id(self) -> None: + update = {"update_id": 1, "callback_query": {"id": "cb-123", "data": "x"}} + assert telegram_callback_query_id(update) == "cb-123" + + def test_returns_none_without_callback_query(self) -> None: + assert telegram_callback_query_id(_message_update(text="hi")) is None + + +class TestTelegramMediaFileId: + def test_picks_largest_photo_by_file_size(self) -> None: + update = _message_update( + photo=[ + {"file_id": "small", "file_size": 100, "width": 90, "height": 90}, + {"file_id": "large", "file_size": 5000, "width": 800, "height": 600}, + {"file_id": "medium", "file_size": 1000, "width": 320, "height": 240}, + ] + ) + assert telegram_media_file_id(update) == ("large", "image/jpeg") + + def test_picks_largest_photo_by_area_when_no_file_size(self) -> None: + update = _message_update( + photo=[ + {"file_id": "small", "width": 90, "height": 90}, + {"file_id": "large", "width": 800, "height": 600}, + ] + ) + assert telegram_media_file_id(update) == ("large", "image/jpeg") + + def test_document_uses_declared_mime_type(self) -> None: + update = _message_update(document={"file_id": "doc1", "mime_type": "application/pdf"}) + assert telegram_media_file_id(update) == ("doc1", "application/pdf") + + def test_document_falls_back_to_default_mime_type(self) -> None: + update = _message_update(document={"file_id": "doc1"}) + assert telegram_media_file_id(update) == ("doc1", "application/octet-stream") + + def test_voice_defaults_to_ogg(self) -> None: + update = _message_update(voice={"file_id": "v1"}) + assert telegram_media_file_id(update) == ("v1", "audio/ogg") + + def test_audio_defaults_to_mpeg(self) -> None: + update = _message_update(audio={"file_id": "a1"}) + assert telegram_media_file_id(update) == ("a1", "audio/mpeg") + + def test_video_defaults_to_mp4(self) -> None: + update = _message_update(video={"file_id": "vi1"}) + assert telegram_media_file_id(update) == ("vi1", "video/mp4") + + def test_accepts_bare_message_object(self) -> None: + message = {"chat": {"id": 1}, "document": {"file_id": "doc1", "mime_type": "text/plain"}} + assert telegram_media_file_id(message) == ("doc1", "text/plain") + + def test_returns_none_without_media(self) -> None: + assert telegram_media_file_id(_message_update(text="hi")) is None + + def test_reads_media_from_callback_query_message(self) -> None: + update = { + "update_id": 1, + "callback_query": { + "id": "cb1", + "data": "x", + "message": {"chat": {"id": 1}, "document": {"file_id": "doc1", "mime_type": "text/plain"}}, + }, + } + assert telegram_media_file_id(update) == ("doc1", "text/plain") + + +class TestTelegramToRun: + async def test_message_text_becomes_user_message(self) -> None: + run = await telegram_to_run(_message_update(text="hello")) + messages = cast("list[Message]", run["messages"]) + assert len(messages) == 1 + assert messages[0].role == "user" + assert messages[0].text == "hello" + assert run["stream"] is False + + async def test_stream_flag_is_forwarded(self) -> None: + run = await telegram_to_run(_message_update(text="hello"), stream=True) + assert run["stream"] is True + + async def test_edited_message_becomes_user_message(self) -> None: + update = {"update_id": 1, "edited_message": {"chat": {"id": 1}, "text": "edited"}} + run = await telegram_to_run(update) + messages = cast("list[Message]", run["messages"]) + assert messages[0].text == "edited" + + async def test_callback_query_data_becomes_user_text(self) -> None: + update = {"update_id": 1, "callback_query": {"id": "cb1", "data": "action:confirm"}} + run = await telegram_to_run(update) + messages = cast("list[Message]", run["messages"]) + assert messages[0].text == "action:confirm" + + async def test_caption_used_when_no_text(self) -> None: + update = _message_update(caption="a cute cat", photo=[{"file_id": "p1", "file_size": 10}]) + run = await telegram_to_run(update) + messages = cast("list[Message]", run["messages"]) + assert messages[0].text == "a cute cat" + + async def test_media_resolved_via_resolver(self) -> None: + update = _message_update(caption="a cute cat", photo=[{"file_id": "p1", "file_size": 10}]) + + async def resolve_file_url(file_id: str) -> str | None: + assert file_id == "p1" + return "https://example.com/p1.jpg" + + run = await telegram_to_run(update, resolve_file_url=resolve_file_url) + messages = cast("list[Message]", run["messages"]) + uris = [c.uri for c in messages[0].contents if c.type == "uri"] + assert uris == ["https://example.com/p1.jpg"] + assert messages[0].text == "a cute cat" + + async def test_media_without_resolver_preserves_text(self) -> None: + update = _message_update(caption="a cute cat", photo=[{"file_id": "p1", "file_size": 10}]) + run = await telegram_to_run(update) + messages = cast("list[Message]", run["messages"]) + assert messages[0].text == "a cute cat" + assert not any(c.type == "uri" for c in messages[0].contents) + + async def test_media_resolver_returning_none_preserves_text(self) -> None: + update = _message_update(caption="a cute cat", photo=[{"file_id": "p1", "file_size": 10}]) + + async def resolve_file_url(file_id: str) -> str | None: + return None + + run = await telegram_to_run(update, resolve_file_url=resolve_file_url) + messages = cast("list[Message]", run["messages"]) + assert messages[0].text == "a cute cat" + + async def test_media_only_unresolved_raises(self) -> None: + update = _message_update(photo=[{"file_id": "p1", "file_size": 10}]) + with pytest.raises(ValueError, match="Cannot resolve"): + await telegram_to_run(update) + + async def test_media_only_no_resolver_raises(self) -> None: + update = _message_update(document={"file_id": "d1"}) + with pytest.raises(ValueError, match="Cannot resolve"): + await telegram_to_run(update) + + async def test_empty_message_raises(self) -> None: + update = {"update_id": 1, "message": {"chat": {"id": 1}}} + with pytest.raises(ValueError, match="no text, caption, or resolvable media"): + await telegram_to_run(update) + + async def test_callback_query_without_data_raises(self) -> None: + update = {"update_id": 1, "callback_query": {"id": "cb1"}} + with pytest.raises(ValueError, match="no actionable"): + await telegram_to_run(update) + + async def test_update_without_actionable_content_raises(self) -> None: + with pytest.raises(ValueError, match="no actionable"): + await telegram_to_run({"update_id": 1, "poll": {"id": "p1"}}) diff --git a/python/packages/hosting-telegram/tests/hosting_telegram/test_rendering.py b/python/packages/hosting-telegram/tests/hosting_telegram/test_rendering.py new file mode 100644 index 00000000000..44110912a14 --- /dev/null +++ b/python/packages/hosting-telegram/tests/hosting_telegram/test_rendering.py @@ -0,0 +1,226 @@ +# Copyright (c) Microsoft. All rights reserved. + +"""Tests for the Telegram Bot API rendering helpers.""" + +from __future__ import annotations + +from collections.abc import AsyncIterator, Sequence + +import pytest +from agent_framework import AgentResponse, AgentResponseUpdate, Content, Message, ResponseStream + +from agent_framework_hosting_telegram import ( + TELEGRAM_MAX_TEXT_LENGTH, + telegram_from_run, + telegram_from_streaming_run, +) + + +def _text_response(text: str) -> AgentResponse[None]: + return AgentResponse(messages=Message(role="assistant", contents=[Content.from_text(text=text)])) + + +class TestTelegramFromRun: + def test_text_response_renders_send_message(self) -> None: + operation = telegram_from_run(_text_response("hello there"), chat_id=555) + assert operation["method"] == "sendMessage" + assert operation["payload"] == {"chat_id": 555, "text": "hello there"} + + def test_parse_mode_is_included_when_given(self) -> None: + operation = telegram_from_run(_text_response("hi"), chat_id=555, parse_mode="MarkdownV2") + assert operation["payload"]["parse_mode"] == "MarkdownV2" + + def test_no_text_or_image_falls_back_to_no_response(self) -> None: + empty = AgentResponse(messages=Message(role="assistant", contents=[])) + operation = telegram_from_run(empty, chat_id=1) + assert operation["method"] == "sendMessage" + assert operation["payload"]["text"] == "(no response)" + + def test_text_is_truncated_to_max_length(self) -> None: + long_text = "x" * (TELEGRAM_MAX_TEXT_LENGTH + 500) + operation = telegram_from_run(_text_response(long_text), chat_id=1) + assert len(operation["payload"]["text"]) == TELEGRAM_MAX_TEXT_LENGTH + assert operation["payload"]["text"] == "x" * TELEGRAM_MAX_TEXT_LENGTH + + def test_image_uri_renders_send_photo(self) -> None: + result = AgentResponse( + messages=Message( + role="assistant", + contents=[Content.from_uri(uri="https://example.com/cat.png", media_type="image/png")], + ) + ) + operation = telegram_from_run(result, chat_id=1) + assert operation["method"] == "sendPhoto" + assert operation["payload"] == {"chat_id": 1, "photo": "https://example.com/cat.png"} + + def test_image_with_text_uses_caption(self) -> None: + result = AgentResponse( + messages=Message( + role="assistant", + contents=[ + Content.from_text(text="a cat"), + Content.from_uri(uri="https://example.com/cat.png", media_type="image/png"), + ], + ) + ) + operation = telegram_from_run(result, chat_id=1, parse_mode="HTML") + assert operation["method"] == "sendPhoto" + assert operation["payload"]["caption"] == "a cat" + assert operation["payload"]["parse_mode"] == "HTML" + + def test_non_image_uri_is_not_rendered_as_photo(self) -> None: + result = AgentResponse( + messages=Message( + role="assistant", + contents=[ + Content.from_text(text="here is a pdf"), + Content.from_uri(uri="https://example.com/report.pdf", media_type="application/pdf"), + ], + ) + ) + operation = telegram_from_run(result, chat_id=1) + assert operation["method"] == "sendMessage" + assert operation["payload"]["text"] == "here is a pdf" + + def test_only_first_image_is_rendered(self) -> None: + result = AgentResponse( + messages=Message( + role="assistant", + contents=[ + Content.from_uri(uri="https://example.com/first.png", media_type="image/png"), + Content.from_uri(uri="https://example.com/second.png", media_type="image/png"), + ], + ) + ) + operation = telegram_from_run(result, chat_id=1) + assert operation["payload"]["photo"] == "https://example.com/first.png" + + +class TestTelegramFromStreamingRun: + async def test_accumulates_text_across_updates(self) -> None: + async def updates() -> AsyncIterator[AgentResponseUpdate]: + yield AgentResponseUpdate(contents=[Content.from_text(text="hel")], role="assistant") + yield AgentResponseUpdate(contents=[Content.from_text(text="lo")], role="assistant") + + stream = ResponseStream(updates(), finalizer=AgentResponse.from_updates) + + operations = [op async for op in telegram_from_streaming_run(stream, chat_id=1, message_id=42)] + + interim = [op for op in operations if op["payload"].get("text") in ("hel", "hello")] + assert operations[0]["method"] == "editMessageText" + assert operations[0]["payload"] == {"chat_id": 1, "message_id": 42, "text": "hel"} + assert operations[1]["payload"]["text"] == "hello" + assert len(interim) == 2 + + async def test_final_edit_includes_parse_mode(self) -> None: + async def updates() -> AsyncIterator[AgentResponseUpdate]: + yield AgentResponseUpdate(contents=[Content.from_text(text="hi")], role="assistant") + + stream = ResponseStream(updates(), finalizer=AgentResponse.from_updates) + + operations = [ + op async for op in telegram_from_streaming_run(stream, chat_id=1, message_id=42, parse_mode="MarkdownV2") + ] + + interim_op = operations[0] + final_op = operations[-1] + assert "parse_mode" not in interim_op["payload"] + assert final_op["method"] == "editMessageText" + assert final_op["payload"]["parse_mode"] == "MarkdownV2" + assert final_op["payload"]["text"] == "hi" + + async def test_edit_matching_initial_placeholder_is_omitted(self) -> None: + async def updates() -> AsyncIterator[AgentResponseUpdate]: + yield AgentResponseUpdate(contents=[Content.from_text(text="...")], role="assistant") + yield AgentResponseUpdate(contents=[Content.from_text(text="done")], role="assistant") + + stream = ResponseStream(updates(), finalizer=AgentResponse.from_updates) + + operations = [ + op + async for op in telegram_from_streaming_run( + stream, + chat_id=1, + message_id=42, + initial_text="...", + ) + ] + + assert len(operations) == 1 + assert operations[0]["payload"]["text"] == "...done" + + async def test_final_text_is_truncated(self) -> None: + long_text = "y" * (TELEGRAM_MAX_TEXT_LENGTH + 200) + + async def updates() -> AsyncIterator[AgentResponseUpdate]: + yield AgentResponseUpdate(contents=[Content.from_text(text=long_text)], role="assistant") + + stream = ResponseStream(updates(), finalizer=AgentResponse.from_updates) + + operations = [op async for op in telegram_from_streaming_run(stream, chat_id=1, message_id=42)] + + assert all(len(op["payload"]["text"]) <= TELEGRAM_MAX_TEXT_LENGTH for op in operations) + + async def test_images_from_final_response_render_as_send_photo(self) -> None: + async def updates() -> AsyncIterator[AgentResponseUpdate]: + yield AgentResponseUpdate( + contents=[Content.from_uri(uri="https://example.com/cat.png", media_type="image/png")], + role="assistant", + ) + + stream = ResponseStream(updates(), finalizer=AgentResponse.from_updates) + + operations = [op async for op in telegram_from_streaming_run(stream, chat_id=1, message_id=42)] + + photo_ops = [op for op in operations if op["method"] == "sendPhoto"] + assert len(photo_ops) == 1 + assert photo_ops[0]["payload"] == {"chat_id": 1, "photo": "https://example.com/cat.png"} + + async def test_no_text_and_no_image_falls_back_to_no_response(self) -> None: + async def updates() -> AsyncIterator[AgentResponseUpdate]: + return + yield # pragma: no cover - make this an async generator with no items + + stream = ResponseStream(updates(), finalizer=AgentResponse.from_updates) + + operations = [op async for op in telegram_from_streaming_run(stream, chat_id=1, message_id=42)] + + assert len(operations) == 1 + assert operations[0]["payload"]["text"] == "(no response)" + + async def test_image_only_final_response_skips_text_edit(self) -> None: + async def updates() -> AsyncIterator[AgentResponseUpdate]: + yield AgentResponseUpdate( + contents=[Content.from_uri(uri="https://example.com/cat.png", media_type="image/png")], + role="assistant", + ) + + stream = ResponseStream(updates(), finalizer=AgentResponse.from_updates) + + operations = [op async for op in telegram_from_streaming_run(stream, chat_id=1, message_id=42)] + + assert operations[0]["method"] == "deleteMessage" + assert operations[0]["payload"] == {"chat_id": 1, "message_id": 42} + assert operations[1]["method"] == "sendPhoto" + + async def test_error_during_iteration_propagates(self) -> None: + async def updates() -> AsyncIterator[AgentResponseUpdate]: + yield AgentResponseUpdate(contents=[Content.from_text(text="partial")], role="assistant") + raise RuntimeError("upstream blew up") + + stream = ResponseStream(updates(), finalizer=AgentResponse.from_updates) + + with pytest.raises(RuntimeError, match="upstream blew up"): + _ = [op async for op in telegram_from_streaming_run(stream, chat_id=1, message_id=42)] + + async def test_error_during_finalization_propagates(self) -> None: + async def updates() -> AsyncIterator[AgentResponseUpdate]: + yield AgentResponseUpdate(contents=[Content.from_text(text="partial")], role="assistant") + + def finalizer(items: Sequence[AgentResponseUpdate]) -> AgentResponse[None]: + raise RuntimeError("finalizer blew up") + + stream = ResponseStream(updates(), finalizer=finalizer) + + with pytest.raises(RuntimeError, match="finalizer blew up"): + _ = [op async for op in telegram_from_streaming_run(stream, chat_id=1, message_id=42)] diff --git a/python/pyproject.toml b/python/pyproject.toml index b7575995643..1d0f9411229 100644 --- a/python/pyproject.toml +++ b/python/pyproject.toml @@ -93,6 +93,7 @@ agent-framework-gemini = { workspace = true } agent-framework-github-copilot = { workspace = true } agent-framework-hosting = { workspace = true } agent-framework-hosting-responses = { workspace = true } +agent-framework-hosting-telegram = { workspace = true } agent-framework-hyperlight = { workspace = true } agent-framework-lab = { workspace = true } agent-framework-mem0 = { workspace = true } diff --git a/python/samples/04-hosting/af-hosting/README.md b/python/samples/04-hosting/af-hosting/README.md index 23ee48cadfd..56901e03610 100644 --- a/python/samples/04-hosting/af-hosting/README.md +++ b/python/samples/04-hosting/af-hosting/README.md @@ -11,9 +11,11 @@ clients, authentication, response construction, and deployment shape. |---|---|---| | [`local_responses/`](./local_responses) | One agent + one `@tool` + native FastAPI route + Responses helper functions + `AgentState` / `SessionStore`. | **Local only.** Start here to learn the helper seam. | | [`local_responses_workflow/`](./local_responses_workflow) | A workflow target behind a native FastAPI route using Responses helper functions, `WorkflowState`, explicit `CheckpointStorage`, and an app-owned checkpoint cursor. | **Local only.** | +| [`local_telegram/`](./local_telegram) | One agent + `aiogram` polling + Telegram conversion helpers + app-owned commands, media policy, and streaming edits. | **Local only.** Requires a Telegram bot token. | -Each sample is self-contained with its own `pyproject.toml`, server `app.py`, -calling script(s), and `storage/` directory. Samples use `[tool.uv.sources]` +Each sample is self-contained with its own `pyproject.toml`, executable +application code, and `storage/` directory. HTTP samples also include calling +scripts. Samples use `[tool.uv.sources]` to wire unreleased hosting packages to the upstream repo while those packages are still pre-PyPI. Once those packages publish, drop the `[tool.uv.sources]` block and let the declared dependencies resolve from PyPI. @@ -27,7 +29,7 @@ Those samples use the Foundry-managed protocol surface with no | Aspect | `af-hosting/` (this directory) | `foundry-hosted-agents/` | |---|---|---| -| Server stack | App-owned FastAPI + hosting protocol helpers | Foundry Hosted Agents runtime | +| Server stack | App-owned framework/native client + hosting protocol helpers | Foundry Hosted Agents runtime | | Protocol surface | The app exposes the route and calls helpers | The platform exposes Responses + Invocations | -| Run target | Local Hypercorn (`local_responses/`, `local_responses_workflow/`) | Hosted Agents or local container targeting the Hosted Agents contract | +| Run target | Local Hypercorn (`local_responses/`, `local_responses_workflow/`) or native polling (`local_telegram/`) | Hosted Agents or local container targeting the Hosted Agents contract | | When to pick this | You need custom hosting code or want to learn the helper seam | You want the Foundry-managed hosting surface | diff --git a/python/samples/04-hosting/af-hosting/local_telegram/README.md b/python/samples/04-hosting/af-hosting/local_telegram/README.md new file mode 100644 index 00000000000..d6d344ada8b --- /dev/null +++ b/python/samples/04-hosting/af-hosting/local_telegram/README.md @@ -0,0 +1,143 @@ +# local_telegram - Telegram helpers with aiogram polling or webhooks + +A local Telegram bot built from the helper-first hosting pieces: + +- an actual Foundry-backed Agent Framework `Agent`; +- `AgentState` and `InMemoryHistoryProvider` for process-local per-chat + continuity; +- `telegram_to_run(...)` for Telegram update to AF conversion; +- `telegram_from_streaming_run(...)` for AF stream to Telegram edit payloads; +- `aiogram` for typed updates, polling/webhook dispatch, file download, and Bot + API calls. + +There is no Telegram client, polling runtime, webhook router, command registry, +or delivery framework in `agent-framework-hosting-telegram`. This sample uses +the native `aiogram` SDK for those concerns. The helpers we provide through +`agent-framework-hosting-telegram` are deliberately agnostic to which SDK you use, +you can just as easily build your telegram bot with other SDKs for python, +see the [Telegram docs](https://core.telegram.org/bots/samples#python) for more + +Each entry point is intentionally self-contained so it can be read and copied +without following a shared sample helper module. Both handle text, captions, +supported media, callback-query data, and the commands `/start`, `/help`, +`/new`, and `/weather `. + +## Run with polling + +Create a Telegram bot with BotFather, then configure: + +```bash +export FOUNDRY_PROJECT_ENDPOINT=https://.services.ai.azure.com +export FOUNDRY_MODEL=gpt-5-nano +export TELEGRAM_BOT_TOKEN=... +az login + +uv run polling_app.py +``` + +The sample asks `aiogram` to clear any existing webhook before polling because +Telegram does not allow polling while a webhook is registered. + +## Run with a webhook + +Configure the public HTTPS URL that Telegram should call and a random secret +used to authenticate webhook deliveries: + +```bash +export FOUNDRY_PROJECT_ENDPOINT=https://.services.ai.azure.com +export FOUNDRY_MODEL=gpt-5-nano +export TELEGRAM_BOT_TOKEN=... +export TELEGRAM_WEBHOOK_URL=https:///telegram/webhook +export TELEGRAM_WEBHOOK_SECRET= +az login + +uv run app.py +``` + +Each entry point declares its complete Agent Framework and third-party +dependency set using PEP 723 inline script metadata, so `uv` creates the +appropriate environment directly from the selected script. + +`app.py` derives its FastAPI route from the path in +`TELEGRAM_WEBHOOK_URL`, registers that URL with Telegram during application +startup, and validates `X-Telegram-Bot-Api-Secret-Token` with +`TELEGRAM_WEBHOOK_SECRET` before accepting an update. + +The app intentionally leaves the webhook registered during shutdown. Deleting +it can race a rolling deployment and remove the webhook that the replacement +process just registered. + +## Behavior to notice + +- **Session continuity:** `telegram_session_id(..., bot_id=bot.id)` follows + Telegram's native identity boundaries. Private chats use + `telegram::`; groups and supergroups use + `telegram::`, creating a shared session for that group. + Including `bot.id` prevents two bots from accidentally sharing state. + aiogram derives that numeric bot id from `TELEGRAM_BOT_TOKEN`, so these local + apps do not need a separate `TELEGRAM_BOT_ID` setting. +- **Starting over:** `/new` calls `state.session_store.delete(session_id)`. + The command itself does not run the agent. On the next ordinary message, + `get_or_create_session(...)` finds no stored value and creates a fresh + `AgentSession` with empty `InMemoryHistoryProvider` state. In a group, this + resets the shared group session; an app that wants per-user group sessions + should include both the chat id and sender id in its app-owned key. +- **Process restarts:** history is intentionally process-local in this sample. + Restarting either app also starts fresh. A durable deployment must replace + both the in-memory session store and history provider deliberately. +- **Commands:** recognized commands are handled by application code and bypass + the agent. Unknown slash commands fall through as ordinary agent input. +- **Callback queries:** the app acknowledges callback queries first to clear + Telegram's loading indicator, then treats callback data as user input unless + it matched an app-owned command. +- **Media:** files larger than 5 MiB are not forwarded. Downloaded media is + converted to an inline data URI so a token-bearing Telegram file URL is not + disclosed to the model provider. If media cannot be resolved, a caption can + still be used as text; unresolved media-only updates are ignored. +- **Streaming:** the app sends a placeholder, applies cumulative text with + `editMessageText`, and throttles edits. Telegram can normalize distinct + payloads to the same rendered content; the app treats only its resulting + `message is not modified` edit error as an idempotent success. Other Bot API + errors still propagate. For an image-only result, the helper deletes the + placeholder before emitting `sendPhoto`. +- **Transport ordering:** polling uses `tasks_concurrency_limit=1`, so this + compact sample processes updates serially. The webhook acknowledges first + and processes in a FastAPI background task; a production app must add its + own per-chat ordering and durable queue if required. +- **Webhook trust:** `TELEGRAM_WEBHOOK_SECRET` authenticates delivery from + Telegram. It does not authorize the Telegram user or chat to access + application data. + +## What the app owns + +The helper package only converts protocol values. `aiogram`, `app.py`, and +`polling_app.py` own: + +- polling or FastAPI webhook setup and update dispatch; +- bounded media download and inline-data conversion, avoiding disclosure of + token-bearing Telegram file URLs to the model provider; +- slash-command policy and session reset; +- native send, photo, typing, callback acknowledgement, and edit calls; +- edit throttling and error logging. + +That code remains visible so an application can replace it with a webhook, +queue, or retry policy without changing the AF conversion helpers. + +## Production readiness + +These are compact hosting samples, not complete production Telegram +deployments. +Before deploying this pattern: + +- use HTTPS and keep webhook secret validation enabled; +- store `TELEGRAM_BOT_TOKEN` in a secret manager and avoid logging Bot API URLs; +- authorize users before mapping a chat id to sensitive/shared state; +- replace process-local history/session state with durable storage partitioned + by tenant/user and define retention; +- make update processing idempotent and preserve per-chat ordering; +- handle Telegram `429` responses and `retry_after` values; +- add bounded retries, delivery telemetry, and dead-letter handling; +- decide how partial streaming edits should recover when a final edit fails. + +> This sample is **self-hosted**. The multi-protocol Telegram + Invocations +> Foundry-hosted sample remains part of the separate Invocations work. diff --git a/python/samples/04-hosting/af-hosting/local_telegram/app.py b/python/samples/04-hosting/af-hosting/local_telegram/app.py new file mode 100644 index 00000000000..eaf2e78268d --- /dev/null +++ b/python/samples/04-hosting/af-hosting/local_telegram/app.py @@ -0,0 +1,303 @@ +# /// script +# requires-python = ">=3.10" +# dependencies = [ +# "agent-framework-core", +# "agent-framework-foundry", +# "agent-framework-hosting", +# "agent-framework-hosting-telegram", +# "aiogram>=3.29.1,<4", +# "azure-identity", +# "fastapi>=0.115.0,<0.138.1", +# "hypercorn>=0.17", +# ] +# /// +# Run with: uv run app.py + +# Copyright (c) Microsoft. All rights reserved. + +"""Run a self-contained Telegram bot as an authenticated webhook. + +This entry point uses native FastAPI routes and aiogram update handling. At +startup it registers ``TELEGRAM_WEBHOOK_URL`` with Telegram and configures +Telegram to send ``TELEGRAM_WEBHOOK_SECRET`` in the +``X-Telegram-Bot-Api-Secret-Token`` header. The route validates that header +before accepting an update. + +Production readiness +--- +The secret header authenticates Telegram's webhook delivery, but it does not +authorize the end user represented by an update. Production apps must still +authorize chat/user ids before loading sensitive state, use durable state and +idempotent update processing, and add bounded background work with delivery +telemetry. See ``README.md#production-readiness``. + +Required environment variables: ``FOUNDRY_PROJECT_ENDPOINT``, +``FOUNDRY_MODEL``, ``TELEGRAM_BOT_TOKEN``, ``TELEGRAM_WEBHOOK_URL``, and +``TELEGRAM_WEBHOOK_SECRET``. + +Run:: + + uv run app.py +""" + +from __future__ import annotations + +import asyncio +import base64 +import hmac +import logging +import os +import time +from collections.abc import AsyncIterator, Mapping +from contextlib import asynccontextmanager +from io import BytesIO +from typing import Annotated, Any, cast +from urllib.parse import urlparse + +from agent_framework import Agent, InMemoryHistoryProvider, ResponseStream, tool +from agent_framework_foundry import FoundryChatClient +from agent_framework_hosting import AgentState +from agent_framework_hosting_telegram import ( + TelegramOperation, + telegram_callback_query_id, + telegram_chat_id, + telegram_command, + telegram_from_streaming_run, + telegram_session_id, + telegram_to_run, +) +from aiogram import Bot, Dispatcher +from aiogram.exceptions import TelegramBadRequest +from aiogram.methods import DeleteMessage, EditMessageText, SendMessage, SendPhoto +from aiogram.types import CallbackQuery, Message, Update +from azure.identity.aio import DefaultAzureCredential +from fastapi import BackgroundTasks, FastAPI, HTTPException, Request, Response +from hypercorn.asyncio import serve +from hypercorn.config import Config + +LOGGER = logging.getLogger(__name__) +EDIT_INTERVAL_SECONDS = 0.4 +MAX_MEDIA_BYTES = 5 * 1024 * 1024 +PLACEHOLDER_TEXT = "..." +ALLOWED_UPDATES = ["message", "edited_message", "callback_query"] +WEBHOOK_URL = os.environ["TELEGRAM_WEBHOOK_URL"] +WEBHOOK_SECRET = os.environ["TELEGRAM_WEBHOOK_SECRET"] +WEBHOOK_PATH = urlparse(WEBHOOK_URL).path or "/" + + +@tool(approval_mode="never_require") +def lookup_weather( + location: Annotated[str, "The city to look up weather for."], +) -> str: + """Return a deterministic weather report for a city.""" + high_temp = 5 + (sum(location.encode("utf-8")) % 21) + reports = { + "Seattle": f"Seattle is rainy with a high of {high_temp}°C.", + "Amsterdam": f"Amsterdam is cloudy with a high of {high_temp}°C.", + "Tokyo": f"Tokyo is clear with a high of {high_temp}°C.", + } + return reports.get(location, f"{location} is sunny with a high of {high_temp}°C.") + + +def create_agent() -> Agent: + """Create the sample weather agent.""" + return Agent( + client=FoundryChatClient(credential=DefaultAzureCredential()), + name="WeatherAgent", + instructions=( + "You are a friendly weather assistant. Use the lookup_weather tool " + "for weather questions and answer in one short sentence." + ), + tools=[lookup_weather], + context_providers=[InMemoryHistoryProvider()], + default_options={"store": False}, + ) + + +state = AgentState(create_agent) +dispatcher = Dispatcher(disable_fsm=True) +bot = Bot(token=os.environ["TELEGRAM_BOT_TOKEN"]) + + +def telegram_update(event_name: str, event: Message | CallbackQuery) -> dict[str, Any]: + """Wrap one aiogram event in the Bot API update shape expected by helpers.""" + return {event_name: event.model_dump(mode="json", by_alias=True, exclude_none=True)} + + +async def execute_operation(operation: TelegramOperation) -> Any: + """Execute one operation produced by a Telegram rendering helper.""" + try: + match operation["method"]: + case "sendMessage": + return await bot(SendMessage.model_validate(operation["payload"])) + case "sendPhoto": + return await bot(SendPhoto.model_validate(operation["payload"])) + case "editMessageText": + return await bot(EditMessageText.model_validate(operation["payload"])) + case "deleteMessage": + return await bot(DeleteMessage.model_validate(operation["payload"])) + case method: + raise ValueError(f"Unsupported Telegram operation: {method}") + except TelegramBadRequest as exc: + if operation["method"] == "editMessageText" and "message is not modified" in exc.message.lower(): + LOGGER.debug("Telegram ignored an edit whose rendered content was unchanged") + return None + raise + + +async def handle_command(update: Mapping[str, Any], command: str) -> bool: + """Handle sample-owned commands and return whether one matched.""" + chat_id = telegram_chat_id(update) + session_id = telegram_session_id(update, bot_id=bot.id) + if chat_id is None or session_id is None: + return False + + name, _, argument = command.partition(" ") + if name == "/start": + text = "Hi! I am a weather assistant. Try asking about a city or use /weather ." + elif name == "/help": + text = "/new - reset this chat\n/weather - look up weather directly\n/help - show this message" + elif name == "/new": + # SessionStore maps the stable Telegram chat key to its current + # AgentSession. Deleting that entry makes the next message create a + # fresh AgentSession with empty in-memory history. + await state.session_store.delete(session_id) + text = "New session started. Your next message begins with empty history." + elif name == "/weather": + text = lookup_weather(location=argument.strip() or "Seattle") + else: + return False + + await bot.send_message(chat_id=chat_id, text=text) + return True + + +async def handle_update(update: Mapping[str, Any]) -> None: + """Process one Telegram update through the sample agent.""" + callback_query_id = telegram_callback_query_id(update) + if callback_query_id is not None: + await bot.answer_callback_query(callback_query_id=callback_query_id) + + if (command := telegram_command(update)) is not None and await handle_command(update, command): + return + + chat_id = telegram_chat_id(update) + session_id = telegram_session_id(update, bot_id=bot.id) + if chat_id is None or session_id is None: + return + + async def resolve_file_url(file_id: str) -> str | None: + file = await bot.get_file(file_id) + if file.file_path is None or (file.file_size is not None and file.file_size > MAX_MEDIA_BYTES): + return None + destination = BytesIO() + await bot.download_file(file.file_path, destination=destination) + encoded = base64.b64encode(destination.getvalue()).decode("ascii") + return f"data:application/octet-stream;base64,{encoded}" + + try: + run = await telegram_to_run(update, resolve_file_url=resolve_file_url, stream=True) + except ValueError: + LOGGER.debug("Ignoring non-actionable Telegram update", exc_info=True) + return + + await bot.send_chat_action(chat_id=chat_id, action="typing") + placeholder = await bot.send_message(chat_id=chat_id, text=PLACEHOLDER_TEXT) + + target = await state.get_target() + # Reuse one AgentSession per Telegram chat. The /new command removes this + # mapping so get_or_create_session creates a clean session next time. + session = await state.get_or_create_session(session_id) + stream = target.run( + run["messages"], + stream=True, + session=session, + options=run["options"], + ) + if not isinstance(stream, ResponseStream): + raise RuntimeError("agent did not return a response stream") + + last_edit_at = 0.0 + async for operation in telegram_from_streaming_run( + stream, + chat_id=chat_id, + message_id=placeholder.message_id, + initial_text=PLACEHOLDER_TEXT, + ): + if operation["method"] == "editMessageText": + delay = EDIT_INTERVAL_SECONDS - (time.monotonic() - last_edit_at) + if delay > 0: + await asyncio.sleep(delay) + last_edit_at = time.monotonic() + await execute_operation(operation) + + # Persist the updated AgentSession back under the stable per-chat key after + # streaming has finalized and the history provider has recorded the turn. + await state.set_session(session_id, session) + + +@dispatcher.message() +async def on_message(message: Message) -> None: + """Handle a new Telegram message.""" + await handle_update(telegram_update("message", message)) + + +@dispatcher.edited_message() +async def on_edited_message(message: Message) -> None: + """Handle an edited Telegram message.""" + await handle_update(telegram_update("edited_message", message)) + + +@dispatcher.callback_query() +async def on_callback_query(callback_query: CallbackQuery) -> None: + """Handle an inline-button callback query.""" + await handle_update(telegram_update("callback_query", callback_query)) + + +@asynccontextmanager +async def lifespan(_: FastAPI) -> AsyncIterator[None]: + """Register the Telegram webhook and close the bot session on shutdown.""" + await bot.set_webhook( + url=WEBHOOK_URL, + secret_token=WEBHOOK_SECRET, + allowed_updates=ALLOWED_UPDATES, + ) + yield + # Leave the webhook registered. Deleting it during rolling shutdown can + # remove the webhook just registered by the replacement process. + await bot.session.close() + + +app = FastAPI(lifespan=lifespan) + + +@app.post(WEBHOOK_PATH, response_model=None) +async def telegram_webhook(request: Request, background_tasks: BackgroundTasks) -> Response: + """Authenticate and enqueue one Telegram webhook update.""" + received_secret = request.headers.get("x-telegram-bot-api-secret-token", "") + if not hmac.compare_digest(received_secret, WEBHOOK_SECRET): + raise HTTPException(status_code=401, detail="invalid Telegram webhook secret") + + try: + payload = cast("dict[str, Any]", await request.json()) + update = Update.model_validate(payload, context={"bot": bot}) + except (TypeError, ValueError) as exc: + raise HTTPException(status_code=400, detail="invalid Telegram update") from exc + + background_tasks.add_task(dispatcher.feed_update, bot, update) + return Response(status_code=200) + + +async def main() -> None: + """Run the webhook sample with Hypercorn for local development.""" + logging.basicConfig(level=logging.INFO) + config = Config() + config.bind = [f"0.0.0.0:{int(os.environ.get('PORT', '8000'))}"] + await serve(cast(Any, app), config) + + +if __name__ == "__main__": + asyncio.run(main()) + +# Sample response: +# HTTP/1.1 200 OK diff --git a/python/samples/04-hosting/af-hosting/local_telegram/polling_app.py b/python/samples/04-hosting/af-hosting/local_telegram/polling_app.py new file mode 100644 index 00000000000..1ba1d537565 --- /dev/null +++ b/python/samples/04-hosting/af-hosting/local_telegram/polling_app.py @@ -0,0 +1,248 @@ +# /// script +# requires-python = ">=3.10" +# dependencies = [ +# "agent-framework-core", +# "agent-framework-foundry", +# "agent-framework-hosting", +# "agent-framework-hosting-telegram", +# "aiogram>=3.29.1,<4", +# "azure-identity", +# ] +# /// +# Run with: uv run polling_app.py + +# Copyright (c) Microsoft. All rights reserved. + +"""Run a self-contained Telegram bot with aiogram long polling. + +Required environment variables: ``FOUNDRY_PROJECT_ENDPOINT``, +``FOUNDRY_MODEL``, and ``TELEGRAM_BOT_TOKEN``. + +Run:: + + az login + uv run polling_app.py +""" + +from __future__ import annotations + +import asyncio +import base64 +import logging +import os +import time +from collections.abc import Mapping +from io import BytesIO +from typing import Annotated, Any + +from agent_framework import Agent, InMemoryHistoryProvider, ResponseStream, tool +from agent_framework_foundry import FoundryChatClient +from agent_framework_hosting import AgentState +from agent_framework_hosting_telegram import ( + TelegramOperation, + telegram_callback_query_id, + telegram_chat_id, + telegram_command, + telegram_from_streaming_run, + telegram_session_id, + telegram_to_run, +) +from aiogram import Bot, Dispatcher +from aiogram.exceptions import TelegramBadRequest +from aiogram.methods import DeleteMessage, EditMessageText, SendMessage, SendPhoto +from aiogram.types import CallbackQuery, Message +from azure.identity.aio import DefaultAzureCredential + +LOGGER = logging.getLogger(__name__) +EDIT_INTERVAL_SECONDS = 0.4 +MAX_MEDIA_BYTES = 5 * 1024 * 1024 +PLACEHOLDER_TEXT = "..." +ALLOWED_UPDATES = ["message", "edited_message", "callback_query"] + + +@tool(approval_mode="never_require") +def lookup_weather( + location: Annotated[str, "The city to look up weather for."], +) -> str: + """Return a deterministic weather report for a city.""" + high_temp = 5 + (sum(location.encode("utf-8")) % 21) + reports = { + "Seattle": f"Seattle is rainy with a high of {high_temp}°C.", + "Amsterdam": f"Amsterdam is cloudy with a high of {high_temp}°C.", + "Tokyo": f"Tokyo is clear with a high of {high_temp}°C.", + } + return reports.get(location, f"{location} is sunny with a high of {high_temp}°C.") + + +def create_agent() -> Agent: + """Create the sample weather agent.""" + return Agent( + client=FoundryChatClient(credential=DefaultAzureCredential()), + name="WeatherAgent", + instructions=( + "You are a friendly weather assistant. Use the lookup_weather tool " + "for weather questions and answer in one short sentence." + ), + tools=[lookup_weather], + context_providers=[InMemoryHistoryProvider()], + default_options={"store": False}, + ) + + +state = AgentState(create_agent) +dispatcher = Dispatcher(disable_fsm=True) + + +def telegram_update(event_name: str, event: Message | CallbackQuery) -> dict[str, Any]: + """Wrap one aiogram event in the Bot API update shape expected by helpers.""" + return {event_name: event.model_dump(mode="json", by_alias=True, exclude_none=True)} + + +async def execute_operation(bot: Bot, operation: TelegramOperation) -> Any: + """Execute one operation produced by a Telegram rendering helper.""" + try: + match operation["method"]: + case "sendMessage": + return await bot(SendMessage.model_validate(operation["payload"])) + case "sendPhoto": + return await bot(SendPhoto.model_validate(operation["payload"])) + case "editMessageText": + return await bot(EditMessageText.model_validate(operation["payload"])) + case "deleteMessage": + return await bot(DeleteMessage.model_validate(operation["payload"])) + case method: + raise ValueError(f"Unsupported Telegram operation: {method}") + except TelegramBadRequest as exc: + if operation["method"] == "editMessageText" and "message is not modified" in exc.message.lower(): + LOGGER.debug("Telegram ignored an edit whose rendered content was unchanged") + return None + raise + + +async def handle_command(bot: Bot, update: Mapping[str, Any], command: str) -> bool: + """Handle sample-owned commands and return whether one matched.""" + chat_id = telegram_chat_id(update) + session_id = telegram_session_id(update, bot_id=bot.id) + if chat_id is None or session_id is None: + return False + + name, _, argument = command.partition(" ") + if name == "/start": + text = "Hi! I am a weather assistant. Try asking about a city or use /weather ." + elif name == "/help": + text = "/new - reset this chat\n/weather - look up weather directly\n/help - show this message" + elif name == "/new": + # SessionStore maps the stable Telegram chat key to its current + # AgentSession. Deleting that entry makes the next message create a + # fresh AgentSession with empty in-memory history. + await state.session_store.delete(session_id) + text = "New session started. Your next message begins with empty history." + elif name == "/weather": + text = lookup_weather(location=argument.strip() or "Seattle") + else: + return False + + await bot.send_message(chat_id=chat_id, text=text) + return True + + +async def handle_update(bot: Bot, update: Mapping[str, Any]) -> None: + """Process one Telegram update through the sample agent.""" + callback_query_id = telegram_callback_query_id(update) + if callback_query_id is not None: + await bot.answer_callback_query(callback_query_id=callback_query_id) + + if (command := telegram_command(update)) is not None and await handle_command(bot, update, command): + return + + chat_id = telegram_chat_id(update) + session_id = telegram_session_id(update, bot_id=bot.id) + if chat_id is None or session_id is None: + return + + async def resolve_file_url(file_id: str) -> str | None: + file = await bot.get_file(file_id) + if file.file_path is None or (file.file_size is not None and file.file_size > MAX_MEDIA_BYTES): + return None + destination = BytesIO() + await bot.download_file(file.file_path, destination=destination) + encoded = base64.b64encode(destination.getvalue()).decode("ascii") + return f"data:application/octet-stream;base64,{encoded}" + + try: + run = await telegram_to_run(update, resolve_file_url=resolve_file_url, stream=True) + except ValueError: + LOGGER.debug("Ignoring non-actionable Telegram update", exc_info=True) + return + + await bot.send_chat_action(chat_id=chat_id, action="typing") + placeholder = await bot.send_message(chat_id=chat_id, text=PLACEHOLDER_TEXT) + + target = await state.get_target() + # Reuse one AgentSession per Telegram chat. The /new command removes this + # mapping so get_or_create_session creates a clean session next time. + session = await state.get_or_create_session(session_id) + stream = target.run( + run["messages"], + stream=True, + session=session, + options=run["options"], + ) + if not isinstance(stream, ResponseStream): + raise RuntimeError("agent did not return a response stream") + + last_edit_at = 0.0 + async for operation in telegram_from_streaming_run( + stream, + chat_id=chat_id, + message_id=placeholder.message_id, + initial_text=PLACEHOLDER_TEXT, + ): + if operation["method"] == "editMessageText": + delay = EDIT_INTERVAL_SECONDS - (time.monotonic() - last_edit_at) + if delay > 0: + await asyncio.sleep(delay) + last_edit_at = time.monotonic() + await execute_operation(bot, operation) + + # Persist the updated AgentSession back under the stable per-chat key after + # streaming has finalized and the history provider has recorded the turn. + await state.set_session(session_id, session) + + +@dispatcher.message() +async def on_message(message: Message, bot: Bot) -> None: + """Handle a new Telegram message.""" + await handle_update(bot, telegram_update("message", message)) + + +@dispatcher.edited_message() +async def on_edited_message(message: Message, bot: Bot) -> None: + """Handle an edited Telegram message.""" + await handle_update(bot, telegram_update("edited_message", message)) + + +@dispatcher.callback_query() +async def on_callback_query(callback_query: CallbackQuery, bot: Bot) -> None: + """Handle an inline-button callback query.""" + await handle_update(bot, telegram_update("callback_query", callback_query)) + + +async def main() -> None: + """Start aiogram long polling until the process is stopped.""" + logging.basicConfig(level=logging.INFO) + bot = Bot(token=os.environ["TELEGRAM_BOT_TOKEN"]) + await bot.delete_webhook(drop_pending_updates=False) + await dispatcher.start_polling( + bot, + allowed_updates=ALLOWED_UPDATES, + tasks_concurrency_limit=1, + ) + + +if __name__ == "__main__": + asyncio.run(main()) + +# Sample output in Telegram: +# User: What is the weather in Tokyo? +# Bot: Tokyo is clear with a high of 18°C. diff --git a/python/samples/04-hosting/af-hosting/local_telegram/pyproject.toml b/python/samples/04-hosting/af-hosting/local_telegram/pyproject.toml new file mode 100644 index 00000000000..988ae3cad4d --- /dev/null +++ b/python/samples/04-hosting/af-hosting/local_telegram/pyproject.toml @@ -0,0 +1,21 @@ +[project] +name = "agent-framework-hosting-sample-local-telegram" +version = "0.0.1" +description = "Local Telegram bot using Agent Framework hosting protocol helpers." +requires-python = ">=3.10" +dependencies = [ + "agent-framework-foundry", + "agent-framework-hosting", + "agent-framework-hosting-telegram", + "aiogram>=3.29.1,<4", + "azure-identity", + "fastapi>=0.115.0,<0.138.1", + "hypercorn>=0.17", +] + +[tool.uv] +package = false + +[tool.uv.sources] +agent-framework-hosting = { git = "https://github.com/microsoft/agent-framework.git", branch = "main", subdirectory = "python/packages/hosting" } +agent-framework-hosting-telegram = { git = "https://github.com/microsoft/agent-framework.git", branch = "main", subdirectory = "python/packages/hosting-telegram" } diff --git a/python/samples/AGENTS.md b/python/samples/AGENTS.md index 6bc8581ed47..ff47976794d 100644 --- a/python/samples/AGENTS.md +++ b/python/samples/AGENTS.md @@ -64,6 +64,11 @@ python/samples/ 4. **Single-file for 01-03**: Only 04-hosting and 05-end-to-end use multi-file projects with their own README. +5. **Self-contained alternatives**: When a sample offers alternative entry + points (for example polling and webhook hosting), keep each entry point + self-contained. Do not extract a shared helper module solely to remove + duplication between sample variants. + ## Default provider All canonical samples (01-get-started) use **Azure AI Foundry project-backed chat** via `FoundryChatClient` diff --git a/python/uv.lock b/python/uv.lock index b15b290e312..6b2ef6f0460 100644 --- a/python/uv.lock +++ b/python/uv.lock @@ -52,6 +52,7 @@ members = [ "agent-framework-github-copilot", "agent-framework-hosting", "agent-framework-hosting-responses", + "agent-framework-hosting-telegram", "agent-framework-hyperlight", "agent-framework-lab", "agent-framework-mem0", @@ -683,6 +684,21 @@ dev = [ { name = "httpx", specifier = ">=0.28.1" }, ] +[[package]] +name = "agent-framework-hosting-telegram" +version = "1.0.0a260709" +source = { editable = "packages/hosting-telegram" } +dependencies = [ + { name = "agent-framework-core", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" }, + { name = "agent-framework-hosting", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" }, +] + +[package.metadata] +requires-dist = [ + { name = "agent-framework-core", editable = "packages/core" }, + { name = "agent-framework-hosting", editable = "packages/hosting" }, +] + [[package]] name = "agent-framework-hyperlight" version = "1.0.0b260709" From a5fbfb8fbc04d9a1a376af44b0dc4053891e7897 Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Fri, 10 Jul 2026 14:28:57 +0200 Subject: [PATCH 2/3] Python: Exclude Telegram samples from aggregate typing Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 5d6987cd-1b67-4ba1-8b54-3c50da6e7607 --- python/pyrightconfig.samples.json | 1 + python/pyrightconfig.samples.py310.json | 1 + python/scripts/workspace_poe_tasks.py | 1 + 3 files changed, 3 insertions(+) diff --git a/python/pyrightconfig.samples.json b/python/pyrightconfig.samples.json index 3d0c80919ee..c231b2fb703 100644 --- a/python/pyrightconfig.samples.json +++ b/python/pyrightconfig.samples.json @@ -8,6 +8,7 @@ "**/_to_delete/**", "**/05-end-to-end/**", "**/harness/**", + "**/local_telegram/**", "**/agent_with_foundry_tracing.py", "**/azure_responses_client_with_foundry.py" ], diff --git a/python/pyrightconfig.samples.py310.json b/python/pyrightconfig.samples.py310.json index 48e9f0a14a7..70741bca033 100644 --- a/python/pyrightconfig.samples.py310.json +++ b/python/pyrightconfig.samples.py310.json @@ -8,6 +8,7 @@ "**/_to_delete/**", "**/05-end-to-end/**", "**/harness/**", + "**/local_telegram/**", "**/agent_with_foundry_tracing.py", "**/azure_responses_client_with_foundry.py", "**/github_copilot/**" diff --git a/python/scripts/workspace_poe_tasks.py b/python/scripts/workspace_poe_tasks.py index 3b3b5a33226..8581b26b5a9 100644 --- a/python/scripts/workspace_poe_tasks.py +++ b/python/scripts/workspace_poe_tasks.py @@ -350,6 +350,7 @@ def _gating_checker_args() -> list[str]: "_to_delete", "05-end-to-end", "harness", + "local_telegram", ) From 17f45129e81adf89a5c3e037575b047143f555ab Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Fri, 10 Jul 2026 14:43:18 +0200 Subject: [PATCH 3/3] Python: Address Telegram helper review feedback Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 5d6987cd-1b67-4ba1-8b54-3c50da6e7607 --- .../_rendering.py | 13 +++++---- .../tests/hosting_telegram/test_rendering.py | 29 +++++++++++++++++++ .../af-hosting/local_telegram/README.md | 8 ++--- .../af-hosting/local_telegram/app.py | 5 +++- .../af-hosting/local_telegram/polling_app.py | 5 +++- 5 files changed, 49 insertions(+), 11 deletions(-) diff --git a/python/packages/hosting-telegram/agent_framework_hosting_telegram/_rendering.py b/python/packages/hosting-telegram/agent_framework_hosting_telegram/_rendering.py index 711225ac868..30d1769096b 100644 --- a/python/packages/hosting-telegram/agent_framework_hosting_telegram/_rendering.py +++ b/python/packages/hosting-telegram/agent_framework_hosting_telegram/_rendering.py @@ -18,8 +18,6 @@ # Telegram's documented maximum length, in UTF-16 code units, for message # text (`sendMessage` / `editMessageText`) and photo captions (`sendPhoto`). -# Agent Framework text is plain `str`; truncating by Python character count -# is a conservative (never-too-long) approximation of the UTF-16 limit. TELEGRAM_MAX_TEXT_LENGTH = 4096 TELEGRAM_MAX_CAPTION_LENGTH = 1024 @@ -39,8 +37,13 @@ class TelegramOperation(TypedDict): def _truncate(text: str, max_length: int) -> str: - """Deterministically cap ``text`` at ``max_length`` characters.""" - return text if len(text) <= max_length else text[:max_length] + """Deterministically cap ``text`` at ``max_length`` UTF-16 code units.""" + units = 0 + for index, char in enumerate(text): + units += 2 if ord(char) > 0xFFFF else 1 + if units > max_length: + return text[:index] + return text def _text_and_image_uris(result: AgentResponse[Any]) -> tuple[str, list[str]]: @@ -49,7 +52,7 @@ def _text_and_image_uris(result: AgentResponse[Any]) -> tuple[str, list[str]]: content.uri for message in result.messages for content in message.contents - if content.type in ("data", "uri") and content.uri and (content.media_type or "").startswith("image/") + if content.type == "uri" and content.uri and (content.media_type or "").startswith("image/") ] return result.text, image_uris diff --git a/python/packages/hosting-telegram/tests/hosting_telegram/test_rendering.py b/python/packages/hosting-telegram/tests/hosting_telegram/test_rendering.py index 44110912a14..049bcc822b4 100644 --- a/python/packages/hosting-telegram/tests/hosting_telegram/test_rendering.py +++ b/python/packages/hosting-telegram/tests/hosting_telegram/test_rendering.py @@ -10,6 +10,7 @@ from agent_framework import AgentResponse, AgentResponseUpdate, Content, Message, ResponseStream from agent_framework_hosting_telegram import ( + TELEGRAM_MAX_CAPTION_LENGTH, TELEGRAM_MAX_TEXT_LENGTH, telegram_from_run, telegram_from_streaming_run, @@ -42,6 +43,10 @@ def test_text_is_truncated_to_max_length(self) -> None: assert len(operation["payload"]["text"]) == TELEGRAM_MAX_TEXT_LENGTH assert operation["payload"]["text"] == "x" * TELEGRAM_MAX_TEXT_LENGTH + def test_text_is_truncated_by_utf16_code_units(self) -> None: + operation = telegram_from_run(_text_response("😀" * (TELEGRAM_MAX_TEXT_LENGTH // 2 + 1)), chat_id=1) + assert operation["payload"]["text"] == "😀" * (TELEGRAM_MAX_TEXT_LENGTH // 2) + def test_image_uri_renders_send_photo(self) -> None: result = AgentResponse( messages=Message( @@ -68,6 +73,30 @@ def test_image_with_text_uses_caption(self) -> None: assert operation["payload"]["caption"] == "a cat" assert operation["payload"]["parse_mode"] == "HTML" + def test_image_caption_is_truncated_by_utf16_code_units(self) -> None: + caption = "😀" * (TELEGRAM_MAX_CAPTION_LENGTH // 2 + 1) + result = AgentResponse( + messages=Message( + role="assistant", + contents=[ + Content.from_text(text=caption), + Content.from_uri(uri="https://example.com/cat.png", media_type="image/png"), + ], + ) + ) + operation = telegram_from_run(result, chat_id=1) + assert operation["payload"]["caption"] == "😀" * (TELEGRAM_MAX_CAPTION_LENGTH // 2) + + def test_inline_image_data_is_not_rendered_as_photo(self) -> None: + result = AgentResponse( + messages=Message( + role="assistant", + contents=[Content.from_data(data=b"image", media_type="image/png")], + ) + ) + operation = telegram_from_run(result, chat_id=1) + assert operation == {"method": "sendMessage", "payload": {"chat_id": 1, "text": "(no response)"}} + def test_non_image_uri_is_not_rendered_as_photo(self) -> None: result = AgentResponse( messages=Message( diff --git a/python/samples/04-hosting/af-hosting/local_telegram/README.md b/python/samples/04-hosting/af-hosting/local_telegram/README.md index d6d344ada8b..c8473a29430 100644 --- a/python/samples/04-hosting/af-hosting/local_telegram/README.md +++ b/python/samples/04-hosting/af-hosting/local_telegram/README.md @@ -12,10 +12,10 @@ A local Telegram bot built from the helper-first hosting pieces: There is no Telegram client, polling runtime, webhook router, command registry, or delivery framework in `agent-framework-hosting-telegram`. This sample uses -the native `aiogram` SDK for those concerns. The helpers we provide through -`agent-framework-hosting-telegram` are deliberately agnostic to which SDK you use, -you can just as easily build your telegram bot with other SDKs for python, -see the [Telegram docs](https://core.telegram.org/bots/samples#python) for more +the native `aiogram` SDK for those concerns. The helpers are deliberately +agnostic to the Telegram SDK you choose. See the +[Telegram documentation](https://core.telegram.org/bots/samples#python) for +other Python SDK options. Each entry point is intentionally self-contained so it can be read and copied without following a shared sample helper module. Both handle text, captions, diff --git a/python/samples/04-hosting/af-hosting/local_telegram/app.py b/python/samples/04-hosting/af-hosting/local_telegram/app.py index eaf2e78268d..c3b9c737678 100644 --- a/python/samples/04-hosting/af-hosting/local_telegram/app.py +++ b/python/samples/04-hosting/af-hosting/local_telegram/app.py @@ -192,7 +192,10 @@ async def resolve_file_url(file_id: str) -> str | None: return None destination = BytesIO() await bot.download_file(file.file_path, destination=destination) - encoded = base64.b64encode(destination.getvalue()).decode("ascii") + data = destination.getvalue() + if len(data) > MAX_MEDIA_BYTES: + return None + encoded = base64.b64encode(data).decode("ascii") return f"data:application/octet-stream;base64,{encoded}" try: diff --git a/python/samples/04-hosting/af-hosting/local_telegram/polling_app.py b/python/samples/04-hosting/af-hosting/local_telegram/polling_app.py index 1ba1d537565..7d65575c9bb 100644 --- a/python/samples/04-hosting/af-hosting/local_telegram/polling_app.py +++ b/python/samples/04-hosting/af-hosting/local_telegram/polling_app.py @@ -166,7 +166,10 @@ async def resolve_file_url(file_id: str) -> str | None: return None destination = BytesIO() await bot.download_file(file.file_path, destination=destination) - encoded = base64.b64encode(destination.getvalue()).decode("ascii") + data = destination.getvalue() + if len(data) > MAX_MEDIA_BYTES: + return None + encoded = base64.b64encode(data).decode("ascii") return f"data:application/octet-stream;base64,{encoded}" try: