A pure Python, asynchronous interface for the Telegram Bot API with comprehensive wrapper and high-level framework for building sophisticated Telegram bots
—
Handlers process specific types of updates from Telegram. The library provides comprehensive handlers for all update types, with filtering capabilities and conversation management.
Foundation for all handler types with filtering and callback execution.
class BaseHandler:
def __init__(self, callback: callable, block: bool = True): ...
def check_update(self, update: Update) -> bool | object: ...
async def handle_update(self, update: Update, application: 'Application', check_result: object, context: CallbackContext) -> object: ...
callback: callable
block: boolHandle text messages, media, and other message types with comprehensive filtering.
class MessageHandler(BaseHandler):
def __init__(
self,
filters: BaseFilter,
callback: callable,
block: bool = True
): ...
filters: BaseFilter
class CommandHandler(BaseHandler):
def __init__(
self,
command: str | list[str],
callback: callable,
filters: BaseFilter = None,
block: bool = True
): ...
command: str | frozenset[str]
filters: BaseFilter | NoneUsage examples:
from telegram.ext import MessageHandler, CommandHandler, filters
# Handle text messages
async def echo(update, context):
await update.message.reply_text(update.message.text)
text_handler = MessageHandler(filters.TEXT & ~filters.COMMAND, echo)
# Handle commands
async def start(update, context):
await update.message.reply_text("Welcome!")
start_handler = CommandHandler("start", start)
# Command with arguments
async def help_command(update, context):
command_args = context.args # List of arguments after command
if command_args:
await update.message.reply_text(f"Help for: {' '.join(command_args)}")
else:
await update.message.reply_text("General help")
help_handler = CommandHandler("help", help_command)
# Handle multiple commands
multi_handler = CommandHandler(["info", "about"], info_callback)Handle callback queries from inline keyboard button presses.
class CallbackQueryHandler(BaseHandler):
def __init__(
self,
callback: callable,
pattern: str | re.Pattern = None,
block: bool = True
): ...
pattern: str | re.Pattern | NoneUsage example:
from telegram.ext import CallbackQueryHandler
import re
async def button_callback(update, context):
query = update.callback_query
await query.answer() # Acknowledge the callback
if query.data == "option_1":
await query.edit_message_text("You chose option 1")
elif query.data == "option_2":
await query.edit_message_text("You chose option 2")
# Handle specific callback data
button_handler = CallbackQueryHandler(button_callback, pattern="^option_")
# Handle all callback queries
all_callbacks = CallbackQueryHandler(button_callback)Handle inline queries when users type @bot_username in any chat.
class InlineQueryHandler(BaseHandler):
def __init__(
self,
callback: callable,
pattern: str | re.Pattern = None,
block: bool = True,
chat_types: list[str] = None
): ...
pattern: str | re.Pattern | None
chat_types: list[str] | NoneUsage example:
from telegram.ext import InlineQueryHandler
from telegram import InlineQueryResultArticle, InputTextMessageContent
async def inline_query(update, context):
query = update.inline_query.query
if not query:
return
results = [
InlineQueryResultArticle(
id="1",
title=f"Result for '{query}'",
input_message_content=InputTextMessageContent(f"You searched for: {query}")
)
]
await update.inline_query.answer(results, cache_time=0)
inline_handler = InlineQueryHandler(inline_query)Manage multi-step conversations with state tracking.
class ConversationHandler(BaseHandler):
def __init__(
self,
entry_points: list[BaseHandler],
states: dict[object, list[BaseHandler]],
fallbacks: list[BaseHandler],
allow_reentry: bool = False,
per_chat: bool = True,
per_user: bool = True,
per_message: bool = False,
conversation_timeout: float = None,
name: str = None,
persistent: bool = False,
map_to_parent: dict = None,
block: bool = True
): ...
END: int = -1
TIMEOUT: int = -2
WAITING: int = -3
entry_points: list[BaseHandler]
states: dict[object, list[BaseHandler]]
fallbacks: list[BaseHandler]Usage example:
from telegram.ext import ConversationHandler, CommandHandler, MessageHandler, filters
# Conversation states
CHOOSING, TYPING_REPLY = range(2)
async def start_conv(update, context):
await update.message.reply_text("What's your name?")
return TYPING_REPLY
async def received_name(update, context):
user_data = context.user_data
user_data['name'] = update.message.text
await update.message.reply_text(f"Nice to meet you, {update.message.text}!")
return ConversationHandler.END
async def cancel(update, context):
await update.message.reply_text("Conversation cancelled.")
return ConversationHandler.END
# Create conversation handler
conv_handler = ConversationHandler(
entry_points=[CommandHandler('start', start_conv)],
states={
TYPING_REPLY: [MessageHandler(filters.TEXT & ~filters.COMMAND, received_name)],
},
fallbacks=[CommandHandler('cancel', cancel)],
)Handlers for specific Telegram features and update types.
class PollHandler(BaseHandler):
def __init__(self, callback: callable, block: bool = True): ...
class PollAnswerHandler(BaseHandler):
def __init__(self, callback: callable, block: bool = True): ...
class ChatMemberHandler(BaseHandler):
def __init__(self, callback: callable, chat_member_types: int = None, block: bool = True): ...
MY_CHAT_MEMBER: int = 1
CHAT_MEMBER: int = 2
ANY_CHAT_MEMBER: int = -1
class ChatJoinRequestHandler(BaseHandler):
def __init__(self, callback: callable, block: bool = True): ...
class ChosenInlineResultHandler(BaseHandler):
def __init__(self, callback: callable, pattern: str | re.Pattern = None, block: bool = True): ...
class ShippingQueryHandler(BaseHandler):
def __init__(self, callback: callable, block: bool = True): ...
class PreCheckoutQueryHandler(BaseHandler):
def __init__(self, callback: callable, block: bool = True): ...
class BusinessConnectionHandler(BaseHandler):
def __init__(self, callback: callable, block: bool = True): ...
class BusinessMessagesDeletedHandler(BaseHandler):
def __init__(self, callback: callable, block: bool = True): ...
class MessageReactionHandler(BaseHandler):
def __init__(self, callback: callable, block: bool = True): ...
class ChatBoostHandler(BaseHandler):
def __init__(self, callback: callable, block: bool = True): ...
class PaidMediaPurchasedHandler(BaseHandler):
def __init__(self, callback: callable, block: bool = True): ...Handle string-based updates for non-Telegram sources.
class StringCommandHandler(BaseHandler):
def __init__(
self,
command: str,
callback: callable,
block: bool = True
): ...
class StringRegexHandler(BaseHandler):
def __init__(
self,
pattern: str | re.Pattern,
callback: callable,
block: bool = True
): ...Handle updates based on their type.
class TypeHandler(BaseHandler):
def __init__(
self,
type: type,
callback: callable,
strict: bool = False,
block: bool = True
): ...
type: type
strict: boolUsage example:
from telegram.ext import TypeHandler
from telegram import Document
async def handle_document(update, context):
document = update.message.document
await update.message.reply_text(f"Received document: {document.file_name}")
# Handle any update containing a Document
doc_handler = TypeHandler(Document, handle_document)Handle messages that start with specific prefixes.
class PrefixHandler(BaseHandler):
def __init__(
self,
prefix: str | list[str],
command: str | list[str],
callback: callable,
filters: BaseFilter = None,
block: bool = True
): ...
prefix: frozenset[str]
command: frozenset[str]
filters: BaseFilter | NoneUsage example:
from telegram.ext import PrefixHandler
async def custom_command(update, context):
command = context.matches[0].group(1) # The command part
await update.message.reply_text(f"Custom command: {command}")
# Handle messages like "!help", "!info", "#status"
prefix_handler = PrefixHandler(['!', '#'], ['help', 'info', 'status'], custom_command)# Handlers are processed in groups (lower numbers first)
application.add_handler(urgent_handler, group=0) # Processed first
application.add_handler(normal_handler, group=1) # Processed second
application.add_handler(fallback_handler, group=2) # Processed last
# Within a group, handlers are processed in order of addition
# First matching handler stops processing (unless handler returns None)# Callback queries with regex patterns
async def pattern_callback(update, context):
match = context.match # The regex match object
data = match.group(1) # Extract captured groups
await update.callback_query.answer(f"Matched: {data}")
pattern_handler = CallbackQueryHandler(pattern_callback, pattern=r"^action_(.+)$")
# Inline queries with patterns
inline_pattern = InlineQueryHandler(inline_callback, pattern=r"^search (.+)")async def safe_handler(update, context):
try:
# Handler logic
await risky_operation()
except Exception as e:
await update.message.reply_text("An error occurred!")
# Re-raise to trigger error handlers if needed
raise
# Global error handling
async def error_handler(update, context):
print(f"Error: {context.error}")
application.add_error_handler(error_handler)from typing import Any, Pattern, Union, List, Dict, Optional, Callable
import re
BaseFilter = Any # Defined in filters moduleInstall with Tessl CLI
npx tessl i tessl/pypi-python-telegram-bot