Telegram Advanced - Upsonic AI

Core Components

TelegramInterface – Wraps an Upsonic Agent for Telegram via FastAPI and Telegram Bot API.
TelegramTools – Used internally for sending messages, files, chat actions, and webhook management.
InterfaceManager.serve – Serves the FastAPI app (e.g. Uvicorn).

Initialization Parameters

Parameter	Type	Default	Description
`agent`	`Agent`	Required	Upsonic `Agent` instance.
`bot_token`	`Optional[str]`	`None`	Telegram Bot API token (or `TELEGRAM_BOT_TOKEN`).
`name`	`str`	`"Telegram"`	Interface name.
`mode`	`Union[InterfaceMode, str]`	`InterfaceMode.TASK`	`TASK` or `CHAT`.
`reset_command`	`Optional[str]`	`"/reset"`	Command to reset chat session (CHAT mode). Set `None` to disable.
`storage`	`Optional[Storage]`	`None`	Storage backend for chat sessions (CHAT mode).
`allowed_user_ids`	`Optional[List[int]]`	`None`	Whitelist of Telegram user IDs; `None` = allow all.
`webhook_secret`	`Optional[str]`	`None`	Secret for webhook validation (`X-Telegram-Bot-Api-Secret-Token`). Falls back to `TELEGRAM_WEBHOOK_SECRET` env var if unset.
`webhook_url`	`Optional[str]`	`None`	Base URL for auto-setting webhook on startup.
`parse_mode`	`Optional[str]`	`"HTML"`	Default parse mode: `"HTML"`, `"Markdown"`, `"MarkdownV2"`, or `None`.
`disable_web_page_preview`	`bool`	`False`	Disable link previews in messages.
`disable_notification`	`bool`	`False`	Send messages silently.
`protect_content`	`bool`	`False`	Protect from forwarding/saving.
`reply_in_groups`	`bool`	`True`	Process messages in groups/supergroups.
`reply_in_channels`	`bool`	`False`	Process channel posts.
`process_edited_messages`	`bool`	`False`	Process edited messages.
`process_callback_queries`	`bool`	`True`	Handle inline keyboard callbacks.
`typing_indicator`	`bool`	`True`	Send “typing” chat action before replying.
`max_message_length`	`int`	`4096`	Max message length before splitting.
`stream`	`bool`	`False`	Stream agent responses by progressively editing the message.
`heartbeat_chat_id`	`Optional[int]`	`None`	Explicit Telegram chat ID for heartbeat delivery. Auto-detected from first incoming message if omitted.

Key Methods

Method	Parameters	Return Type	Description
`attach_routes`	None	`APIRouter`	Returns the FastAPI router with Telegram endpoints.
`health_check`	None	`Dict[str, Any]`	Returns health status, configuration (bot connectivity, mode, reset_command, whitelist, parse_mode, behavior flags), and optional bot info from Telegram.
`is_user_allowed`	`user_id: int`	`bool`	Returns whether the user is allowed (whitelist check).

Full Configuration Example

import os
from upsonic import Agent
from upsonic.interfaces import InterfaceManager, TelegramInterface, InterfaceMode

agent = Agent(
    model="anthropic/claude-sonnet-4-6",
    name="TelegramChatBot",
    system_prompt="You are a helpful AI assistant on Telegram.",
    print=True,
    workspace="/my_agent_folder",
)

telegram = TelegramInterface(
    agent=agent,
    name="TelegramChat",
    bot_token=os.getenv("TELEGRAM_BOT_TOKEN"),
    webhook_secret=os.getenv("TELEGRAM_WEBHOOK_SECRET"),
    webhook_url=os.getenv("TELEGRAM_WEBHOOK_URL"),
    mode=InterfaceMode.CHAT,
    reset_command="/reset",
    allowed_user_ids=[1279809673],
    parse_mode="HTML",
    disable_web_page_preview=False,
    disable_notification=False,
    protect_content=False,
    max_message_length=4096,
    reply_in_groups=True,
    reply_in_channels=False,
    process_edited_messages=False,
    process_callback_queries=True,
    typing_indicator=True,
)

manager = InterfaceManager(interfaces=[telegram])
manager.serve(host="0.0.0.0", port=8000, reload=False)

Human-in-the-Loop (HITL) Confirmation

Tools decorated with @tool(requires_confirmation=True) pause the run until the user confirms or rejects in Telegram. The interface sends a message with inline Confirm / Reject buttons; callbacks are handled automatically. Works in both TASK and CHAT modes.

import os
from upsonic import Agent
from upsonic.interfaces import InterfaceManager, TelegramInterface, InterfaceMode
from upsonic.tools import tool

@tool(requires_confirmation=True)
def calculate_factorial(n: int) -> int:
    """Calculate the factorial of a number."""
    if n == 0:
        return 1
    return n * calculate_factorial(n - 1)

agent = Agent(
    model="anthropic/claude-sonnet-4-6",
    tools=[calculate_factorial],
    name="TelegramChatBot",
    system_prompt="You are a friendly AI assistant on Telegram. You remember conversation context.",
    print=True,
)

telegram = TelegramInterface(
    agent=agent,
    bot_token=os.getenv("TELEGRAM_BOT_TOKEN"),
    webhook_url=os.getenv("TELEGRAM_WEBHOOK_URL"),
    webhook_secret=os.getenv("TELEGRAM_WEBHOOK_SECRET"),
    mode=InterfaceMode.CHAT,
    reset_command="/reset",
)

manager = InterfaceManager(interfaces=[telegram])
manager.serve(host="0.0.0.0", port=8000)

Streaming

Set stream=True to progressively edit the Telegram message as tokens arrive from the agent. An initial message is sent immediately and then updated in-place as new chunks are generated. Updates are throttled to ~1 second intervals to respect Telegram rate limits. Works in both TASK and CHAT modes.

import os
from upsonic import Agent
from upsonic.interfaces import InterfaceManager, TelegramInterface

agent = Agent(
    model="anthropic/claude-sonnet-4-6",
    name="TelegramBot",
)

telegram = TelegramInterface(
    agent=agent,
    bot_token=os.getenv("TELEGRAM_BOT_TOKEN"),
    webhook_url=os.getenv("TELEGRAM_WEBHOOK_URL"),
    mode="chat",
    stream=True,
)

manager = InterfaceManager(interfaces=[telegram])

if __name__ == "__main__":
    manager.serve(host="0.0.0.0", port=8000)

Heartbeat

When used with an AutonomousAgent that has heartbeat=True, the interface periodically sends the agent’s heartbeat_message to the agent, then delivers the response to the Telegram chat. The target chat ID is resolved automatically from the first incoming message. You can also set it explicitly via heartbeat_chat_id.

import os
from upsonic import AutonomousAgent
from upsonic.interfaces import InterfaceManager, TelegramInterface

agent = AutonomousAgent(
    model="anthropic/claude-sonnet-4-6",
    name="TelegramBot",
    heartbeat=True,
    heartbeat_period=30,
    heartbeat_message="Summarize any new updates.",
)

telegram = TelegramInterface(
    agent=agent,
    bot_token=os.getenv("TELEGRAM_BOT_TOKEN"),
    webhook_url=os.getenv("TELEGRAM_WEBHOOK_URL"),
    mode="chat",
    # heartbeat_chat_id=1279809673,  # optional explicit override
)

manager = InterfaceManager(interfaces=[telegram])

if __name__ == "__main__":
    manager.serve(host="0.0.0.0", port=8000)

Endpoints

All endpoints are mounted under the /telegram prefix.

`POST /telegram/webhook`

Receives Telegram updates (messages, edited messages, channel posts, callback queries).
Validates X-Telegram-Bot-Api-Secret-Token if webhook_secret is set.
Processes updates in the background; responds with 200 {"ok": true} so Telegram does not retry.

`POST /telegram/set-webhook`

Sets the bot webhook URL.
Query: url (required), secret_token (optional), drop_pending_updates (optional, default False).
Returns {"success": bool}.

If you didn’t set TELEGRAM_WEBHOOK_URL, you can register the webhook manually after starting the server:

curl -X POST 'http://localhost:8000/telegram/set-webhook?url=https://YOUR_NGROK_URL/telegram/webhook'

`POST /telegram/delete-webhook`

Removes the current webhook.
Query: drop_pending_updates (optional).
Returns {"success": bool}.

`GET /telegram/webhook-info`

Returns current webhook status from Telegram (e.g. URL, pending update count).

`GET /telegram/health`

Health and status of the interface: bot connectivity, configuration (mode, whitelist, parse_mode, behavior flags), and optional bot info from Telegram.

curl http://localhost:8000/telegram/health

​Core Components

​Initialization Parameters

​Key Methods

​Full Configuration Example

​Human-in-the-Loop (HITL) Confirmation

​Streaming

​Heartbeat

​Endpoints

​POST /telegram/webhook

​POST /telegram/set-webhook

​POST /telegram/delete-webhook

​GET /telegram/webhook-info

​GET /telegram/health