The fastest way to build a Telegram Bot natively

Building Native Telegram Bots at Lightning Speed: A Technical Deep Dive

Telegram bots have become indispensable tools for automation, notifications, and interactive services. While many frameworks exist, building natively with Telegram's Bot API offers unparalleled performance and control. This guide will walk you through creating a high-performance Telegram bot from scratch.

Understanding Telegram's Bot API Architecture

Telegram's Bot API operates on a simple HTTP-based request-response model with two primary methods:

1. Long Polling: Continuously checks for updates
2. Webhooks: Receives push notifications from Telegram servers

For development speed and simplicity, we'll focus on long polling, though production systems should eventually migrate to webhooks.

Prerequisites

Before diving in, ensure you have:

• A Telegram account
• Python 3.8+ installed
• Basic understanding of HTTP protocols
• requests library installed (pip install requests)

Step 1: Creating Your Bot and Obtaining API Token

1. Start a chat with @botfather
2. Send /newbot command
3. Follow prompts to name your bot and get your API token

import requests

TOKEN = "YOUR_API_TOKEN"
BASE_URL = f"https://api.telegram.org/bot{TOKEN}"

Step 2: Core Bot Infrastructure

We'll implement a robust bot handler with these components:

1. Update fetcher
2. Command parser
3. Response generator
4. State manager

import json
import time
from typing import Dict, Any

class TelegramBot:
    def __init__(self, token: str):
        self.token = token
        self.base_url = f"https://api.telegram.org/bot{token}"
        self.offset = 0  # For update tracking
        self.user_states = {}  # Simple state management

    def _make_request(self, method: str, params: Dict[str, Any] = None) -> Dict:
        """Generic request handler"""
        url = f"{self.base_url}/{method}"
        response = requests.post(url, json=params)
        return response.json()

    def get_updates(self) -> Dict:
        """Fetch new updates with long polling"""
        params = {
            'offset': self.offset + 1 if self.offset else None,
            'timeout': 30  # Long polling timeout
        }
        response = self._make_request('getUpdates', params)
        if response.get('ok') and response.get('result'):
            self.offset = max(update['update_id'] for update in response['result'])
        return response

Step 3: Implementing Command Handlers

Let's create a modular command handling system:

class TelegramBot(TelegramBot):
    def __init__(self, token: str):
        super().__init__(token)
        self.command_handlers = {
            'start': self._handle_start,
            'help': self._handle_help,
            'echo': self._handle_echo
        }

    def _handle_start(self, chat_id: int, _: Dict) -> None:
        """Handle /start command"""
        welcome_msg = "🚀 Bot activated! Type /help for commands."
        self._send_message(chat_id, welcome_msg)

    def _handle_help(self, chat_id: int, _: Dict) -> None:
        """Handle /help command"""
        help_text = """
        Available commands:
        /start - Initialize bot
        /help - Show this message
        /echo [text] - Repeat back text
        """
        self._send_message(chat_id, help_text.strip())

    def _handle_echo(self, chat_id: int, update: Dict) -> None:
        """Handle /echo command"""
        message_text = update['message']['text']
        echo_text = ' '.join(message_text.split()[1:])  # Remove command
        self._send_message(chat_id, echo_text or "Nothing to echo!")

Step 4: Message Processing Pipeline

Here's the core message processing logic:

class TelegramBot(TelegramBot):
    def process_updates(self) -> None:
        """Main processing loop"""
        while True:
            try:
                updates = self.get_updates()
                if updates.get('result'):
                    for update in updates['result']:
                        self._process_single_update(update)
            except Exception as e:
                print(f"Error processing updates: {e}")
                time.sleep(5)  # Backoff on error

    def _process_single_update(self, update: Dict) -> None:
        """Process individual update"""
        if 'message' not in update:
            return

        message = update['message']
        chat_id = message['chat']['id']

        if 'text' in message:
            text = message['text']
            command = text.split()[0].lower() if text.startswith('/') else None

            if command and command[1:] in self.command_handlers:
                self.command_handlers[command[1:]](chat_id, update)
            else:
                self._send_message(chat_id, "Unknown command. Try /help")

Step 5: Sending Messages and Beyond

Implement rich message capabilities:

class TelegramBot(TelegramBot):
    def _send_message(self, chat_id: int, text: str, 
                     parse_mode: str = None) -> Dict:
        """Send text message with optional formatting"""
        params = {
            'chat_id': chat_id,
            'text': text
        }
        if parse_mode in ['MarkdownV2', 'HTML']:
            params['parse_mode'] = parse_mode
        return self._make_request('sendMessage', params)

    def send_photo(self, chat_id: int, photo_url: str, 
                  caption: str = None) -> Dict:
        """Send photo from URL"""
        params = {
            'chat_id': chat_id,
            'photo': photo_url
        }
        if caption:
            params['caption'] = caption
        return self._make_request('sendPhoto', params)

    def send_keyboard(self, chat_id: int, text: str, 
                     buttons: list) -> Dict:
        """Send message with custom keyboard"""
        keyboard = {
            'keyboard': buttons,
            'resize_keyboard': True,
            'one_time_keyboard': True
        }
        params = {
            'chat_id': chat_id,
            'text': text,
            'reply_markup': keyboard
        }
        return self._make_request('sendMessage', params)

Step 6: Running Your Bot

Here's how to start your bot:

if __name__ == "__main__":
    bot = TelegramBot(TOKEN)
    print("Bot started. Press Ctrl+C to stop.")
    try:
        bot.process_updates()
    except KeyboardInterrupt:
        print("\nBot stopped.")

Performance Optimization Techniques

1. Update Batching: Process multiple updates in parallel
2. Local Caching: Store frequent data locally
3. Connection Pooling: Reuse HTTP connections

from concurrent.futures import ThreadPoolExecutor

class HighPerformanceTelegramBot(TelegramBot):
    def __init__(self, token: str, workers: int = 4):
        super().__init__(token)
        self.executor = ThreadPoolExecutor(max_workers=workers)
        self.session = requests.Session()  # Reuse connections

    def _make_request(self, method: str, params: Dict[str, Any] = None) -> Dict:
        """Optimized request handler with connection pooling"""
        url = f"{self.base_url}/{method}"
        with self.session.post(url, json=params) as response:
            return response.json()

    def process_updates(self) -> None:
        """Parallel update processing"""
        while True:
            try:
                updates = self.get_updates()
                if updates.get('result'):
                    list(self.executor.map(
                        self._process_single_update, 
                        updates['result']
                    ))
            except Exception as e:
                print(f"Error processing updates: {e}")
                time.sleep(5)

Advanced Features Implementation

1. Inline Query Support

class TelegramBot(TelegramBot):
    def _process_inline_query(self, update: Dict) -> None:
        """Handle inline queries"""
        inline_query = update['inline_query']
        query_id = inline_query['id']
        query_text = inline_query['query']

        results = [{
            'type': 'article',
            'id': '1',
            'title': 'Sample Result',
            'input_message_content': {
                'message_text': f"You searched: {query_text}"
            }
        }]

        self._make_request('answerInlineQuery', {
            'inline_query_id': query_id,
            'results': json.dumps(results)
        })

2. Webhook Setup (Production Ready)

import ssl
from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/webhook', methods=['POST'])
def webhook():
    update = request.json
    bot.process_webhook_update(update)
    return jsonify({'status': 'ok'})

class TelegramBot(TelegramBot):
    def set_webhook(self, url: str, certificate: str = None) -> Dict:
        """Configure webhook URL"""
        params = {'url': url}
        if certificate:
            params['certificate'] = certificate
        return self._make_request('setWebhook', params)

    def process_webhook_update(self, update: Dict) -> None:
        """Process webhook update"""
        self._process_single_update(update)

Monitoring and Analytics

Implement basic monitoring:

from datetime import datetime
import statistics

class MonitoredTelegramBot(TelegramBot):
    def __init__(self, token: str):
        super().__init__(token)
        self.metrics = {
            'message_count': 0,
            'response_times': [],
            'last_update': None
        }

    def _make_request(self, method: str, params: Dict[str, Any] = None) -> Dict:
        """Track performance metrics"""
        start_time = time.time()
        result = super()._make_request(method, params)
        elapsed = time.time() - start_time

        self.metrics['response_times'].append(elapsed)
        if len(self.metrics['response_times']) > 100:
            self.metrics['response_times'].pop(0)

        return result

    def get_stats(self) -> Dict:
        """Return performance statistics"""
        times = self.metrics['response_times']
        return {
            'total_messages': self.metrics['message_count'],
            'avg_response_time': statistics.mean(times) if times else 0,
            'max_response_time': max(times) if times else 0,
            'last_update': self.metrics['last_update']
        }

Deployment Considerations

1. Containerization: Dockerize your bot for easy deployment
2. Logging: Implement comprehensive logging
3. Rate Limiting: Handle Telegram's rate limits (30 messages/second)

import logging
from tenacity import retry, stop_after_attempt, wait_exponential

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

class ProductionTelegramBot(TelegramBot):
    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=4, max=10)
    )
    def _make_request(self, method: str, params: Dict[str, Any] = None) -> Dict:
        """Retry failed requests with exponential backoff"""
        try:
            response = super()._make_request(method, params)
            if not response.get('ok'):
                logging.error(f"API error: {response.get('description')}")
                raise Exception(response.get('description'))
            return response
        except requests.exceptions.RequestException as e:
            logging.error(f"Request failed: {e}")
            raise

Final Thoughts

This implementation provides a robust foundation for a high-performance Telegram bot. Key takeaways:

1. Native implementation offers better control and performance than frameworks
2. Proper error handling and retry logic are crucial
3. Monitoring is essential for production systems
4. Webhooks should be used for production deployments

To extend this bot, consider adding:

• Database integration for persistent storage
• Natural language processing for advanced interactions
• Payment processing via Telegram Payments
• Integration with other APIs and services

The complete code is available on GitHub (include your repo link here). Happy bot building!

---

🚀 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.