90 MINS intermediate
8. Messaging & Social Bots
Module 08: Messaging Bots
Telegram and Discord Bots from Zero to Production
Messaging bots are the most widely deployed form of automation software. A Telegram bot can automate customer support, deliver alerts, run payment flows, and manage communities. A Discord bot can moderate servers, run games, deliver scheduled content, and integrate with external APIs. Both platforms expose rich APIs that make sophisticated automation accessible with relatively simple code.
🤖 Telegram Bot: Complete Implementation
from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup
from telegram.ext import (
Application, CommandHandler, MessageHandler, CallbackQueryHandler,
ConversationHandler, filters, ContextTypes
)
import os
from loguru import logger
TOKEN = os.environ['TELEGRAM_BOT_TOKEN']
# Conversation states
NAME, EMAIL, CONFIRM = range(3)
async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
keyboard = [
[InlineKeyboardButton('📊 Get Report', callback_data='report')],
[InlineKeyboardButton('⚙️ Settings', callback_data='settings')],
[InlineKeyboardButton('💬 Support', callback_data='support')],
]
await update.message.reply_text(
f'Welcome, {update.effective_user.first_name}!\n\nWhat can I do for you today?',
parse_mode='HTML',
reply_markup=InlineKeyboardMarkup(keyboard)
)
async def button_callback(update: Update, context: ContextTypes.DEFAULT_TYPE):
query = update.callback_query
await query.answer() # removes loading state
if query.data == 'report':
await query.edit_message_text('Generating your report...⏳')
# fetch data, generate report
await query.edit_message_text('✅ Report ready:\n\n• Items: 1,247\n• Revenue: $48,392\n• Errors: 3')
elif query.data == 'settings':
await query.edit_message_text('⚙️ Settings coming soon.')
# Multi-step conversation flow
async def register_start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int:
await update.message.reply_text('📝 What is your name?')
return NAME
async def register_name(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int:
context.user_data['name'] = update.message.text
await update.message.reply_text(f'Nice to meet you, {update.message.text}! What is your email?')
return EMAIL
async def register_email(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int:
context.user_data['email'] = update.message.text
name = context.user_data['name']
email = context.user_data['email']
keyboard = [[InlineKeyboardButton('✅ Confirm', callback_data='confirm_yes'),
InlineKeyboardButton('❌ Cancel', callback_data='confirm_no')]]
await update.message.reply_text(
f'Confirm registration?\n\nName: {name}\nEmail: {email}',
reply_markup=InlineKeyboardMarkup(keyboard)
)
return CONFIRM
async def error_handler(update: object, context: ContextTypes.DEFAULT_TYPE):
logger.error(f'Update {update} caused error {context.error}')
def main():
app = Application.builder().token(TOKEN).build()
# Register handlers
app.add_handler(CommandHandler('start', start))
app.add_handler(CallbackQueryHandler(button_callback))
# Conversation handler for registration flow
conv_handler = ConversationHandler(
entry_points=[CommandHandler('register', register_start)],
states={
NAME: [MessageHandler(filters.TEXT & ~filters.COMMAND, register_name)],
EMAIL: [MessageHandler(filters.TEXT & ~filters.COMMAND, register_email)],
CONFIRM: [CallbackQueryHandler(button_callback)],
},
fallbacks=[CommandHandler('cancel', lambda u, c: ConversationHandler.END)]
)
app.add_handler(conv_handler)
app.add_error_handler(error_handler)
logger.info('Bot started. Polling...')
app.run_polling(allowed_updates=Update.ALL_TYPES)
if __name__ == '__main__':
main()🎮 Discord Bot: Full Implementation
import discord
from discord.ext import commands, tasks
from discord import app_commands
import os
from loguru import logger
from datetime import datetime
intents = discord.Intents.default()
intents.message_content = True
intents.members = True
class AutomationBot(commands.Bot):
def __init__(self):
super().__init__(command_prefix='!', intents=intents)
async def setup_hook(self):
await self.tree.sync() # sync slash commands
self.scheduled_report.start()
logger.info('Bot setup complete')
async def on_ready(self):
logger.info(f'Logged in as {self.user} (ID: {self.user.id})')
await self.change_presence(activity=discord.Activity(
type=discord.ActivityType.watching,
name='the data stream'
))
bot = AutomationBot()
# Slash command with autocomplete
@bot.tree.command(name='scrape', description='Scrape a URL and return a summary')
@app_commands.describe(url='The URL to scrape', pages='Number of pages')
async def scrape_command(interaction: discord.Interaction, url: str, pages: int = 1):
await interaction.response.defer() # avoid timeout for slow operations
# Do the actual work
result = {'items': 42, 'duration': '3.2s'}
embed = discord.Embed(title='Scrape Complete', color=discord.Color.green())
embed.add_field(name='URL', value=url, inline=False)
embed.add_field(name='Items Found', value=str(result['items']), inline=True)
embed.add_field(name='Duration', value=result['duration'], inline=True)
embed.set_footer(text=f'Requested by {interaction.user.display_name}')
embed.timestamp = datetime.now()
await interaction.followup.send(embed=embed)
@bot.event
async def on_member_join(member: discord.Member):
channel = member.guild.system_channel
if channel:
await channel.send(
f'Welcome to the server, {member.mention}! 🎉\n'
f'Head to <#rules-channel> to get started.'
)
@tasks.loop(hours=24)
async def scheduled_report():
for guild in bot.guilds:
report_channel = discord.utils.get(guild.text_channels, name='bot-reports')
if report_channel:
embed = discord.Embed(title='Daily Automation Report', color=0x00A3FF)
embed.add_field(name='Status', value='All systems nominal ✅')
await report_channel.send(embed=embed)
bot.run(os.environ['DISCORD_TOKEN'])Automation Arena: Chatbot Sandbox
Sandbox initialized. Waiting for bot deployment...
bot_handler.py
Python 3
Sandbox Editor
Knowledge Check
Ready to test your understanding of 8. Messaging & Social Bots?