Integration of Gemini API into Telegram bot: guide for developers

# Asynchronous Integration of Gemini API into a Telegram Bot: A Practical Guide Using aiogram 3.x

When your Telegram bot suddenly stops responding due to a 429 error from the Gemini API, it's not just a bug—it's technical debt. In this article, we break it down step by step: how to properly integrate Gemini into an asynchronous bot on aiogram 3.x to avoid blocking the event loop, work around RPM/TPM limits, and implement function calls without sacrificing performance.

Architecture: From a Simple Handler to a Production Solution

The basic user-bot interaction flow via Gemini looks straightforward: request → processing → response. But real-world scenarios demand extra layers. Calling a synchronous API asynchronously without proper wrapping will freeze the aiogram event loop, especially under heavy load. That's why the key components are:

• Asyncio Queue — for buffering incoming requests
• Rate Limiter — to comply with Gemini limits (5–60 RPM)
• Response Cache — to save tokens on repeated requests
• Streaming Handler — to bypass Telegram timeouts (10 seconds)

The Gemini API enforces strict limits even on paid plans: maximum response latency can hit 15 seconds, and the free tier caps at just 5 requests per minute. Without queues and caching, your user experience will be toast.

Client Setup: Synchronous SDK in an Asynchronous Environment

The Google GenAI SDK remains synchronous despite community hopes. That means calling generate_content() directly in an aiogram handler will block the entire event loop. The fix: wrap calls in asyncio.to_thread or use run_in_executor.

import asyncio
from google import genai

class AsyncGeminiClient:
    def __init__(self, model: str = "gemini-3-flash-preview"):
        self.client = genai.Client()
        self.model = model
    
    async def generate(self, prompt: str) -> str:
        loop = asyncio.get_event_loop()
        response = await loop.run_in_executor(
            None,
            self._sync_generate,
            prompt
        )
        return response
    
    def _sync_generate(self, prompt: str) -> str:
        response = self.client.models.generate_content(
            model=self.model,
            contents=prompt
        )
        return response.text

This is a minimal working wrapper. For production, add exception handling and execution metrics.

Integration with aiogram: From /start to Streaming Responses

A basic handler needs to do more than just accept messages: manage state by showing "typing," split long responses, and cache results. Telegram caps message length at 4096 characters—ignore this and you'll get cutoffs.

@router.message()
async def handle_message(message: types.Message):
    await message.bot.send_chat_action(chat_id=message.chat.id, action="typing")
    
    try:
        response = await gemini.generate(message.text)
        if len(response) > 4000:
            for i in range(0, len(response), 4000):
                await message.answer(response[i:i+4000])
        else:
            await message.answer(response)
    except Exception as e:
        logging.error(f"Gemini error: {e}")
        await message.answer("Chto-then poshlo not so. Poprobuyte pozzhe.")

For long-running requests, use streaming—update a single message as text generates. This dodges timeouts and boosts UX.

Function Calls: When Gemini Becomes an Agent

Function Calling lets the model invoke external APIs based on request context. For example, if a user says "book a meeting for tomorrow at 3 PM," Gemini can parse params and call your schedule_meeting function.

Key steps:

• Describe the function in JSON Schema format—so the model understands its signature.
• Pass the description in GenerateContentConfig as tools.
• Handle function_call in the response and execute the actual call.
• Feed the result back into the chat as part of the conversation.

Example function declaration:

schedule_meeting_function = {
    "name": "schedule_meeting",
    "description": "Withzdayot meetingsu with ukazannymi uchastnikami",
    "parameters": {
        "type": "object",
        "properties": {
            "attendees": {"type": "array", "items": {"type": "string"}},
            "date": {"type": "string"},
            "time": {"type": "string"},
            "topic": {"type": "string"}
        },
        "required": ["attendees", "date", "time", "topic"]
    }
}

This turns a static bot into a dynamic assistant that can interact with external systems.

Caching and Rate Limiting: Protection from Bankruptcy and Bans

Gemini bills by tokens. Without caching, you'll pay for every repeat question. A simple in-memory cache using MD5 hash of the prompt with TTL fixes that:

class SimpleCache:
    def __init__(self, ttl_seconds: int = 3600):
        self._cache = {}
        self._ttl = ttl_seconds
    
    def get(self, key: str) -> Optional[str]:
        # check TTL and vozvrat values
    
    def set(self, key: str, value: str):
        self._cache[key] = (value, datetime.now())
    
    @staticmethod
    def hash_prompt(prompt: str) -> str:
        return hashlib.md5(prompt.lower().strip().encode()).hexdigest()

Rate limiting is essential even on paid tiers. Implement it as an async context manager or middleware:

class AsyncRateLimiter:
    def __init__(self, max_requests: int, time_window: int = 60):
        self.max_requests = max_requests
        self.time_window = time_window
        self.requests = []
        self._lock = asyncio.Lock()
    
    async def acquire(self):
        async with self._lock:
            now = time.time()
            self.requests = [t for t in self.requests if now - t < self.time_window]
            if len(self.requests) >= self.max_requests:
                sleep_time = self.time_window - (now - self.requests[0])
                await asyncio.sleep(sleep_time + 0.1)
                return await self.acquire()
            self.requests.append(now)

Common Mistakes and How to Avoid Them

Three classic pitfalls everyone hits:

• Error 429 (RESOURCE_EXHAUSTED) — triggered by exceeding RPM, TPM, or RPD. Fix: local rate limiter + exponential backoff.
• Telegram Timeouts — the server waits 10 seconds for a response. If Gemini is slow, use streaming or send interim messages.
• Outdated Models — Google regularly sunsets old versions. In March 2026, Pro models went subscription-only. Always verify model availability before deploying.

Key Takeaways:

• The synchronous Gemini SDK needs run_in_executor wrapping—or you'll block the event loop.
• Implement rate limiting on the bot side, don't rely on the API.
• Streaming is the only way around Telegram's 10-second timeout for long responses.
• Caching saves cash and cuts latency on repeats.
• Function Calling evolves your bot from text generator to full agent.

Adopting these practices cuts production incidents and keeps things stable as load grows. Don't cut corners on architecture—save on tokens instead.

— Editorial Team