Free Lead Bot in Telegram with Python and Yandex Cloud
Solo developers and small teams can build a simple CRM using a Telegram bot—zero cost. The bot qualifies incoming leads with just two inline-question prompts, creates a dedicated topic in a supergroup for each client, and enables full two-way message relay. Deploy on Yandex Cloud Functions with free tier: 1 million invocations and 1 GB of Object Storage per month.
Flow: User follows a deep link from the website → bot asks about business type and goal → creates a topic like "Ivan — E-commerce — AI Agent" → all communication is centralized in the group with search and notifications.
Lightweight Wrapper Over Telegram Bot API
For Cloud Functions, lightweight solutions without heavy frameworks like aiogram work best. Cold start is minimized, using pure httpx. Full API layer spans just 57 lines:
"""telegram_api.py — minimal wrapper over Bot API."""
import os
import httpx
BOT_TOKEN = os.environ.get("BOT_TOKEN", "")
BASE_URL = f"https://api.telegram.org/bot{BOT_TOKEN}"
def _call(method: str, **kwargs) -> dict:
resp = httpx.post(f"{BASE_URL}/{method}", json=kwargs, timeout=10)
return resp.json()
def send_message(chat_id, text, reply_markup=None, message_thread_id=None):
params = {"chat_id": chat_id, "text": text, "parse_mode": "HTML"}
if reply_markup:
params["reply_markup"] = reply_markup
if message_thread_id:
params["message_thread_id"] = message_thread_id
return _call("sendMessage", **params)
def create_forum_topic(chat_id, name):
return _call("createForumTopic", chat_id=chat_id, name=name)
def copy_message(chat_id, from_chat_id, message_id, message_thread_id=None):
params = {"chat_id": chat_id, "from_chat_id": from_chat_id, "message_id": message_id}
if message_thread_id:
params["message_thread_id"] = message_thread_id
return _call("copyMessage", **params)copy_message is preferred over forward_message: it removes the "Forwarded from..." label, keeping messages clean.
Webhook Handler with Validation
Yandex Cloud Function directly handles HTTPS requests from Telegram (allow-unauthenticated-invoke). Validation uses X-Telegram-Bot-Api-Secret-Token:
def handler(event, context):
"""Yandex Cloud Function entry point."""
headers = event.get("headers", {})
secret = headers.get("X-Telegram-Bot-Api-Secret-Token") or \
headers.get("x-telegram-bot-api-secret-token", "")
if secret != WEBHOOK_SECRET:
return {"statusCode": 403, "body": "Forbidden"}
body = event.get("body", "{}")
update = json.loads(body) if isinstance(body, str) else body
if "callback_query" in update:
_handle_callback(update["callback_query"])
elif "message" in update:
_handle_message(update["message"])
return {"statusCode": 200, "body": "ok"}Separating logic into _handle_callback and _handle_message keeps the code clean and maintainable.
Stateless Lead Qualification
The /start command triggers a two-question survey: first, business type; second, goal. Callback data encodes both answers: q2:sales:ecommerce. The 64-byte limit is respected for pre-set options.
Example of the first question:
def _handle_start(chat_id, from_user):
if storage.get_topic_id_by_user(chat_id):
tg.send_message(chat_id,
"You've already submitted a request. We'll contact you soon.\n\n"
"If you'd like to add anything, just reply here.")
return
keyboard = {"inline_keyboard": [
[{"text": "E-commerce / Marketplaces", "callback_data": "q1:ecommerce"}],
[{"text": "Services / SaaS", "callback_data": "q1:services"}],
[{"text": "Manufacturing", "callback_data": "q1:production"}],
[{"text": "IT / Startup", "callback_data": "q1:it"}],
[{"text": "Other", "callback_data": "q1:other"}],
]}
tg.send_message(chat_id, f"Hi, {from_user.get('first_name', '')}! I'm VexAI bot.\n\n"
"Answer 2 quick questions, and we’ll reach out.")
tg.send_message(chat_id, "<b>What kind of business do you run?</b>", reply_markup=keyboard)Duplicate prevention: checks mapping in Object Storage. For "Other," a temporary state file is used.
The second question works similarly, passing the q1_key in callback data.
Topic Creation and Message Relay
After the survey, a topic is generated with a 128-character limit:
def _finalize(chat_id, from_user, q1_answer, q2_answer):
tg.send_message(chat_id, "Thanks! We'll contact you shortly.")
full_name = f"{from_user.get('first_name', '')} {from_user.get('last_name', '')}".strip()
topic_title = f"{full_name} — {q1_answer} — {q2_answer}"
if len(topic_title) > 128:
topic_title = topic_title[:125] + "..."
result = tg.create_forum_topic(ADMIN_GROUP_ID, topic_title)
topic_id = result["result"]["message_thread_id"]
summary = (f"<b>New Lead</b>\n\n"
f"<b>Name:</b> {full_name}\n"
f"<b>Business:</b> {q1_answer}\n"
f"<b>Goal:</b> {q2_answer}\n"
f"<b>Source:</b> website")
tg.send_message(ADMIN_GROUP_ID, summary, message_thread_id=topic_id)
storage.save_topic_mapping(topic_id, chat_id)
storage.save_user_mapping(chat_id, topic_id)Message relay in _handle_message:
- Admin replies in the topic → copied to the client.
- Client messages the bot → copied into the topic.
Deep links supported: /start website for source tracking.
Setup and Limitations
- Create a supergroup and enable topics.
- Add the bot as admin: manage topics, send messages.
- Configure Yandex Cloud: service account with
storage.editor, bucket, access keys. - Set up webhook with secret token.
You get:
- Searchable lead history.
- Real-time notifications.
- Pinned lead cards.
Limitations: no sales funnel, analytics, or team collaboration. Ideal for up to 20 leads/month.
Key Takeaways
- Free deployment: Yandex Cloud Functions + Object Storage in free tier.
- Stateless design: callback data holds state; minimal storage needed.
- Full relay: complete conversation in topics—no external tools.
- Security: secret token + duplicate protection.
- Scalability: supports up to 20 leads/month without paid CRM.
— Editorial Team
No comments yet.