The Telegram interface connects your agent to Telegram’s Bot API. It supports both polling (for development) and webhooks (for production), handles photos, voice messages, audio files, and documents, and provides built-in access control.
Setup
1. Create a Bot
Open Telegram and message @BotFather:
- Send
/newbot
- Choose a name and username
- Copy the bot token (e.g.,
123456:ABC-DEF1234ghIkl-zyx57W2v1u123ew11)
2. Set the Token
export TELEGRAM_BOT_TOKEN="123456:ABC-DEF1234ghIkl-zyx57W2v1u123ew11"
3. Run Your Bot
import asyncio
import os
from definable.agents import Agent
from definable.models import OpenAIChat
from definable.interfaces import TelegramInterface, TelegramConfig
agent = Agent(
model=OpenAIChat(id="gpt-4o"),
instructions="You are a helpful Telegram assistant.",
)
config = TelegramConfig(bot_token=os.environ["TELEGRAM_BOT_TOKEN"])
interface = TelegramInterface(agent=agent, config=config)
asyncio.run(interface.serve_forever())
TelegramConfig Reference
Authentication
Telegram Bot API token from BotFather.
Receiver Mode
How the bot receives messages. "polling" for development, "webhook" for production.
Seconds between polling requests (polling mode only).
Long-polling timeout in seconds (polling mode only).
Webhook Settings
Public HTTPS URL for receiving updates. Required when mode="webhook".
webhook_path
str
default:"/webhook/telegram"
URL path for the webhook endpoint.
Port for the webhook HTTP server.
Secret token for verifying webhook requests from Telegram.
Access Control
Only accept messages from these Telegram user IDs. All users allowed if not set.
Only accept messages from these chat IDs. All chats allowed if not set.
Message formatting: "HTML", "MarkdownV2", "Markdown", or None for plain text.
Telegram’s message character limit. Long responses are split automatically.
Timeouts
HTTP connection timeout in seconds.
HTTP request timeout in seconds.
Polling Mode (Development)
Polling is the simplest mode. The bot periodically asks Telegram for new messages. No public URL or HTTPS certificate is needed.
config = TelegramConfig(
bot_token="YOUR_TOKEN",
mode="polling", # default
polling_interval=0.5, # check every 500ms
polling_timeout=30, # long-polling timeout
)
Use polling for local development and testing. It works behind NATs and firewalls with no setup.
Webhook Mode (Production)
Webhooks are more efficient for production. Telegram pushes updates to your server as they arrive — no polling delay.
config = TelegramConfig(
bot_token="YOUR_TOKEN",
mode="webhook",
webhook_url="https://your-domain.com/webhook/telegram",
webhook_port=8443,
webhook_secret="your-secret-token",
)
- A publicly accessible HTTPS URL
- Port 443, 80, 88, or 8443
- Valid SSL certificate (use Let’s Encrypt or a reverse proxy)
The webhook_secret is strongly recommended. It prevents unauthorized requests to your webhook endpoint. Telegram sends this token in the X-Telegram-Bot-Api-Secret-Token header.
| Telegram Type | Converted To | Notes |
|---|
| Photos | Image | Largest resolution is selected |
| Voice messages | Audio | Includes duration and mime type |
| Audio files | Audio | Includes file metadata |
| Documents | File | Includes filename and mime type |
images, audio, and files parameters, so tools and the model can access them.
The agent can return media in its response. Images are sent as photos, and files are sent as documents:
from definable.tools import tool
from definable.media import Image
@tool
def generate_chart(data: str) -> Image:
"""Generate a chart from data."""
chart_path = create_chart(data)
return Image(filepath=chart_path)
Access Control
Restrict who can use the bot:
config = TelegramConfig(
bot_token="YOUR_TOKEN",
allowed_user_ids=[123456789, 987654321], # Only these users
allowed_chat_ids=[111222333], # Only these chats
)
from definable.agents import Agent
from definable.models import OpenAIChat
from definable.tools import tool
@tool
def get_weather(city: str) -> str:
"""Get the current weather for a city."""
return f"Sunny, 24°C in {city}"
@tool
def set_reminder(text: str, minutes: int, _session_state: dict = None) -> str:
"""Set a reminder."""
reminders = _session_state.setdefault("reminders", [])
reminders.append({"text": text, "minutes": minutes})
return f"Reminder set: '{text}' in {minutes} minutes."
agent = Agent(
model=OpenAIChat(id="gpt-4o"),
tools=[get_weather, set_reminder],
instructions="You are a helpful Telegram assistant with weather and reminder tools.",
)
Complete Production Example
import asyncio
import os
import logging
from definable.agents import Agent, AgentConfig, LoggingMiddleware
from definable.models import OpenAIChat
from definable.interfaces import (
TelegramInterface,
TelegramConfig,
LoggingHook,
AllowlistHook,
)
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("telegram_bot")
agent = Agent(
model=OpenAIChat(id="gpt-4o"),
instructions="You are a customer support agent for Acme Corp.",
config=AgentConfig(max_iterations=5),
).use(LoggingMiddleware(logger))
config = TelegramConfig(
bot_token=os.environ["TELEGRAM_BOT_TOKEN"],
mode="webhook",
webhook_url="https://bot.acme.com/webhook/telegram",
webhook_secret=os.environ.get("WEBHOOK_SECRET"),
max_session_history=30,
session_ttl_seconds=1800, # 30-minute sessions
max_concurrent_requests=20,
parse_mode="HTML",
)
interface = (
TelegramInterface(agent=agent, config=config)
.add_hook(LoggingHook())
.add_hook(AllowlistHook(allowed_user_ids={"12345", "67890"}))
)
asyncio.run(interface.serve_forever())
Error Handling
The Telegram interface maps API errors to specific exception types:
| HTTP Status | Exception | Cause |
|---|
| 401 | InterfaceAuthenticationError | Invalid bot token |
| 429 | InterfaceRateLimitError | Telegram rate limit exceeded |
| 400 | InterfaceMessageError | Bad request (invalid chat, etc.) |
| Other | InterfaceConnectionError | Network or server error |
error_message is sent to the user, and all on_error hooks are invoked. 
