Building Native Telegram Bots with Maximum Performance
Telegram bots have become indispensable tools for automation, customer service, and interactive experiences. While many frameworks abstract away the underlying protocols, building natively against Telegram's API offers unparalleled speed and control. This guide will walk you through creating a high-performance Telegram bot using raw API calls with Python.
Understanding Telegram's Bot Architecture
Telegram bots operate through HTTPS requests to Telegram's Bot API endpoints. The two primary methods for receiving updates are:
- Webhooks (Recommended for production)
- Long Polling (Good for development)
We'll implement both approaches to give you complete flexibility.
Prerequisites
Before we begin, ensure you have:
- Python 3.8+ installed
- A Telegram account
- The
requestslibrary installed (pip install requests) - A registered bot with @botfather (save your API token)
Setting Up the Base Bot Class
Let's create a foundational bot class that handles core functionality:
import requests
import time
import json
from threading import Thread
class NativeTelegramBot:
def __init__(self, token):
self.token = token
self.base_url = f"https://api.telegram.org/bot{self.token}/"
self.last_update_id = 0
self.running = False
def _make_request(self, method, params=None, files=None):
url = self.base_url + method
if files:
response = requests.post(url, data=params, files=files)
else:
response = requests.post(url, json=params)
return response.json()
def get_me(self):
return self._make_request("getMe")
def get_updates(self, offset=None, limit=100, timeout=0):
params = {
"timeout": timeout,
"limit": limit
}
if offset:
params["offset"] = offset
return self._make_request("getUpdates", params)
def send_message(self, chat_id, text, parse_mode=None, reply_to_message_id=None):
params = {
"chat_id": chat_id,
"text": text
}
if parse_mode:
params["parse_mode"] = parse_mode
if reply_to_message_id:
params["reply_to_message_id"] = reply_to_message_id
return self._make_request("sendMessage", params)
Implementing Long Polling
Long polling is the simplest way to get started:
class PollingBot(NativeTelegramBot):
def __init__(self, token):
super().__init__(token)
self.handlers = {}
def add_handler(self, update_type, handler):
self.handlers[update_type] = handler
def process_update(self, update):
if "message" in update:
message = update["message"]
if "text" in message:
if "text" in self.handlers:
self.handlers<a href="message">"text"</a>
def start_polling(self, interval=0.1):
self.running = True
print("Bot started polling...")
while self.running:
updates = self.get_updates(
offset=self.last_update_id + 1,
timeout=30
)
if updates.get("ok"):
for update in updates["result"]:
self.last_update_id = update["update_id"]
self.process_update(update)
time.sleep(interval)
def stop_polling(self):
self.running = False
Implementing Webhooks
For production environments, webhooks are more efficient:
from flask import Flask, request, jsonify
class WebhookBot(NativeTelegramBot):
def __init__(self, token, webhook_url, secret_token=None):
super().__init__(token)
self.webhook_url = webhook_url
self.secret_token = secret_token
self.app = Flask(__name__)
self.handlers = {}
@self.app.route('/', methods=['POST'])
def webhook():
if self.secret_token:
if request.headers.get('X-Telegram-Bot-Api-Secret-Token') != self.secret_token:
return jsonify({"status": "unauthorized"}), 403
update = request.get_json()
self.process_update(update)
return jsonify({"status": "ok"})
def set_webhook(self):
params = {
"url": self.webhook_url
}
if self.secret_token:
params["secret_token"] = self.secret_token
return self._make_request("setWebhook", params)
def delete_webhook(self):
return self._make_request("deleteWebhook")
def run(self, host='0.0.0.0', port=5000):
self.app.run(host=host, port=port)
Advanced Features Implementation
Let's add support for more sophisticated bot features:
Inline Keyboard Markup
def create_inline_keyboard(self, buttons):
keyboard = []
for row in buttons:
keyboard_row = []
for button in row:
keyboard_row.append({
"text": button["text"],
"callback_data": button.get("callback_data")
})
keyboard.append(keyboard_row)
return {"inline_keyboard": keyboard}
def send_message_with_keyboard(self, chat_id, text, buttons):
params = {
"chat_id": chat_id,
"text": text,
"reply_markup": self.create_inline_keyboard(buttons)
}
return self._make_request("sendMessage", params)
File Handling
def send_photo(self, chat_id, photo_path, caption=None):
with open(photo_path, 'rb') as photo:
files = {'photo': photo}
params = {'chat_id': chat_id}
if caption:
params['caption'] = caption
return self._make_request("sendPhoto", params, files)
def get_file(self, file_id):
response = self._make_request("getFile", {"file_id": file_id})
if response.get("ok"):
file_path = response["result"]["file_path"]
return f"https://api.telegram.org/file/bot{self.token}/{file_path}"
return None
Performance Optimization Techniques
To maximize your bot's performance:
- Connection Pooling: Reuse HTTP connections
- Bulk Processing: Handle multiple updates at once
- Async Processing: Process updates in parallel
Here's how to implement connection pooling:
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
class OptimizedNativeTelegramBot(NativeTelegramBot):
def __init__(self, token):
super().__init__(token)
self.session = requests.Session()
# Configure retry strategy
retry_strategy = Retry(
total=3,
backoff_factor=1,
status_forcelist=[429, 500, 502, 503, 504]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
self.session.mount("https://", adapter)
def _make_request(self, method, params=None, files=None):
url = self.base_url + method
try:
if files:
response = self.session.post(url, data=params, files=files)
else:
response = self.session.post(url, json=params)
return response.json()
except requests.exceptions.RequestException as e:
print(f"Request failed: {e}")
return {"ok": False, "error": str(e)}
Error Handling and Rate Limiting
Telegram imposes rate limits (30 messages per second to different chats). Implement proper error handling:
def safe_send_message(self, chat_id, text, max_retries=3):
for attempt in range(max_retries):
try:
response = self.send_message(chat_id, text)
if response.get("ok"):
return True
error_code = response.get("error_code")
if error_code == 429: # Rate limited
retry_after = response.get("parameters", {}).get("retry_after", 1)
time.sleep(retry_after)
continue
except Exception as e:
print(f"Attempt {attempt + 1} failed: {str(e)}")
time.sleep(1)
return False
Putting It All Together: Echo Bot Example
Here's a complete working example:
if __name__ == "__main__":
TOKEN = "YOUR_BOT_TOKEN"
# Create and configure bot
bot = PollingBot(TOKEN)
# Add handler for text messages
def handle_message(message):
chat_id = message["chat"]["id"]
text = message.get("text", "")
bot.send_message(chat_id, f"You said: {text}")
bot.add_handler("text", handle_message)
# Start polling
try:
bot.start_polling()
except KeyboardInterrupt:
bot.stop_polling()
print("Bot stopped")
Deployment Considerations
For production deployment:
- Use Webhooks: More efficient than polling
- Add Secret Token: For webhook security
- Implement Queue: For handling high volumes
- Monitor Performance: Track response times
# Production-ready webhook setup
if __name__ == "__main__":
TOKEN = "YOUR_BOT_TOKEN"
WEBHOOK_URL = "https://yourdomain.com/webhook"
SECRET_TOKEN = "your_secure_random_token"
bot = WebhookBot(TOKEN, WEBHOOK_URL, SECRET_TOKEN)
def handle_message(message):
chat_id = message["chat"]["id"]
text = message.get("text", "")
bot.send_message(chat_id, f"Processed: {text}")
bot.add_handler("text", handle_message)
# Set webhook on startup
print(bot.set_webhook())
# Run Flask app
bot.run(port=8443)
Conclusion
Building Telegram bots natively provides maximum control and performance. By implementing the techniques in this guide, you can create bots that:
- Respond in under 100ms
- Handle thousands of requests per minute
- Support all Telegram features
- Scale efficiently
Remember to always:
- Handle rate limits gracefully
- Secure your webhook endpoints
- Monitor your bot's performance
- Keep your dependencies minimal
The native approach gives you the flexibility to optimize every aspect of your bot's operation, making it the fastest way to build high-performance Telegram bots.
🚀 Stop Writing Boilerplate Prompts
If you want to skip the setup and code 10x faster with complete AI architecture patterns, grab my Senior React Developer AI Cookbook ($19). It includes Server Action prompt libraries, UI component generation loops, and hydration debugging strategies.
Browse all 10+ developer products at the Apollo AI Store | Or snipe Solana tokens free via @ApolloSniper_Bot.
Top comments (0)