Python: Add Telegram hosting helpers and samples (#7047)
* Python: Add Telegram hosting helpers Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 5d6987cd-1b67-4ba1-8b54-3c50da6e7607 * 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: Address Telegram helper review feedback Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 5d6987cd-1b67-4ba1-8b54-3c50da6e7607 * Python: Serialize Telegram webhook sessions Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: d76a9c32-d170-426d-a64f-b70958b08b12
This commit is contained in:
Eduard van Valkenburg
committed by
GitHub
GitHub
parent
d93fc2dd74
commit
7ca8bb55b6
@@ -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`.
|
||||
|
||||
@@ -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:<bot_id>:<user_id>`; other chats return
|
||||
`telegram:<bot_id>:<chat_id>`, 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.
|
||||
|
||||
@@ -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` |
|
||||
|
||||
@@ -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
|
||||
@@ -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:<bot_id>:<user_id>`; other chats use
|
||||
`telegram:<bot_id>:<chat_id>`.
|
||||
- `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/).
|
||||
@@ -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",
|
||||
]
|
||||
@@ -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<name>[A-Za-z0-9_]+)(?:@(?P<bot>[A-Za-z0-9_]+))?(?P<rest>.*)$", 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:<bot_id>:<user_id>`` for a private chat,
|
||||
``telegram:<bot_id>:<chat_id>`` 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,
|
||||
)
|
||||
@@ -0,0 +1,171 @@
|
||||
# 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`).
|
||||
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`` 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]]:
|
||||
"""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 == "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})
|
||||
@@ -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"
|
||||
@@ -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"}})
|
||||
@@ -0,0 +1,255 @@
|
||||
# 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_CAPTION_LENGTH,
|
||||
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_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(
|
||||
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_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(
|
||||
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)]
|
||||
@@ -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 }
|
||||
|
||||
@@ -8,6 +8,7 @@
|
||||
"**/_to_delete/**",
|
||||
"**/05-end-to-end/**",
|
||||
"**/harness/**",
|
||||
"**/local_telegram/**",
|
||||
"**/agent_with_foundry_tracing.py",
|
||||
"**/azure_responses_client_with_foundry.py"
|
||||
],
|
||||
|
||||
@@ -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/**"
|
||||
|
||||
@@ -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 |
|
||||
|
||||
@@ -0,0 +1,147 @@
|
||||
# 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 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,
|
||||
supported media, callback-query data, and the commands `/start`, `/help`,
|
||||
`/new`, and `/weather <city>`.
|
||||
|
||||
## Run with polling
|
||||
|
||||
Create a Telegram bot with BotFather, then configure:
|
||||
|
||||
```bash
|
||||
export FOUNDRY_PROJECT_ENDPOINT=https://<your-project>.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://<your-project>.services.ai.azure.com
|
||||
export FOUNDRY_MODEL=gpt-5-nano
|
||||
export TELEGRAM_BOT_TOKEN=...
|
||||
export TELEGRAM_WEBHOOK_URL=https://<your-host>/telegram/webhook
|
||||
export TELEGRAM_WEBHOOK_SECRET=<random-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:<bot_id>:<user_id>`; groups and supergroups use
|
||||
`telegram:<bot_id>:<chat_id>`, 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, but serializes each chat's
|
||||
updates with an in-process lock so `/new` cannot race an in-flight response.
|
||||
A multi-process deployment must instead use its storage backend's locking or
|
||||
transaction mechanisms, or another cross-process ordering strategy.
|
||||
- **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; use
|
||||
storage-backed locking/transactions or another distributed coordination
|
||||
mechanism when running more than one process;
|
||||
- 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.
|
||||
@@ -0,0 +1,312 @@
|
||||
# /// 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. This sample serializes each chat's updates in-process; distributed
|
||||
deployments need their backend storage's locking or transaction mechanisms, or
|
||||
another cross-process ordering strategy. 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"])
|
||||
session_locks: dict[str, asyncio.Lock] = {}
|
||||
|
||||
|
||||
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 <city>."
|
||||
elif name == "/help":
|
||||
text = "/new - reset this chat\n/weather <city> - 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)
|
||||
|
||||
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
|
||||
|
||||
# Background webhook tasks may overlap. Serialize each chat so /new cannot
|
||||
# delete a session while an earlier response is still updating it.
|
||||
async with session_locks.setdefault(session_id, asyncio.Lock()):
|
||||
if (command := telegram_command(update)) is not None and await handle_command(update, command):
|
||||
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)
|
||||
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:
|
||||
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
|
||||
@@ -0,0 +1,251 @@
|
||||
# /// 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 <city>."
|
||||
elif name == "/help":
|
||||
text = "/new - reset this chat\n/weather <city> - 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)
|
||||
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:
|
||||
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.
|
||||
@@ -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" }
|
||||
@@ -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 **Microsoft Foundry project-backed chat** via `FoundryChatClient`
|
||||
|
||||
@@ -350,6 +350,7 @@ SAMPLE_TYPING_EXCLUDES = (
|
||||
"_to_delete",
|
||||
"05-end-to-end",
|
||||
"harness",
|
||||
"local_telegram",
|
||||
)
|
||||
|
||||
|
||||
|
||||
Generated
+16
@@ -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 @@ test = [
|
||||
{ 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"
|
||||
|
||||
Reference in New Issue
Block a user