Telepot helps you build applications for Telegram Bot API.It works on Python 2.7 and Python 3. For Python 3.5+, it also has an async versionbased on asyncio.
For a time, I tried to list the features here like many projects do. Eventually, I gave up.
Common and straight-forward features are too trivial to worth listing.For more unique and novel features, I cannot find standard terms to describe them.The best way to experience telepot is by reading this page and going through theexamples. Let’s go.
- Installation
- Get a token
- Test the account
- Receive messages
- An easier way to receive messages
- Send a message
- Quickly glance a message
- Custom Keyboard and Inline Keyboard
- Message has a Flavor
- Inline Query
- Maintain Threads of Conversation
- Inline Handler per User
- Async Version (Python 3.5+)
- More Examples »
Installation¶
pip:
$ pip install telepot$ pip install telepot --upgrade # UPGRADE
easy_install:
$ easy_install telepot$ easy_install --upgrade telepot # UPGRADE
Get a token¶
To use the Telegram Bot API, you firsthave to get a bot accountby chatting with BotFather.
BotFather will give you a token, something like 123456789:ABCdefGhIJKlmNoPQRsTUVwxyZ
.With the token in hand, you can start using telepot to access the bot account.
Test the account¶
>>> import telepot>>> bot = telepot.Bot('***** PUT YOUR TOKEN HERE *****')>>> bot.getMe(){'first_name': 'Your Bot', 'username': 'YourBot', 'id': 123456789}
Receive messages¶
Bots cannot initiate conversations with users. You have to send it a message first.Get the message by calling Bot.getUpdates():
>>> from pprint import pprint>>> response = bot.getUpdates()>>> pprint(response)[{'message': {'chat': {'first_name': 'Nick', 'id': 999999999, 'type': 'private'}, 'date': 1465283242, 'from': {'first_name': 'Nick', 'id': 999999999}, 'message_id': 10772, 'text': 'Hello'}, 'update_id': 100000000}]
999999999
is obviously a fake id. Nick
is my real name, though.
The chat
field represents the conversation. Its type
can be private
,group
, or channel
(whose meanings should be obvious, I hope). Above,Nick
just sent a private
message to the bot.
According to Bot API, the method getUpdatesreturns an array of Update objects.As you can see, an Update object is nothing more than a Python dictionary.In telepot, Bot API objects are represented as dictionary.
Note the update_id
. It is an ever-increasing number. Next time you should usegetUpdates(offset=100000001)
to avoid getting the same old messages over and over.Giving an offset
essentially acknowledges to the server that you have receivedall update_id
s lower than offset
:
>>> bot.getUpdates(offset=100000001)[]
An easier way to receive messages¶
It is troublesome to keep checking messages while managing offset
. Let telepottake care of the mundane stuff and notify you whenever new messages arrive:
>>> from telepot.loop import MessageLoop>>> def handle(msg):... pprint(msg)...>>> MessageLoop(bot, handle).run_as_thread()
After setting this up, send it a few messages. Sit back and monitor themessages arriving.
Send a message¶
Sooner or later, your bot will want to send you messages. You should havediscovered your own user id from above interactions. I will keep using myfake id of 999999999
. Remember to substitute your own (real) id:
>>> bot.sendMessage(999999999, 'Hey!')
Quickly glance a message¶
When processing a message, a few pieces of information are so central that youalmost always have to extract them. Use telepot.glance() to extract“headline info”. Try this skeleton, a bot which echoes what you said:
import sysimport timeimport telepotfrom telepot.loop import MessageLoopdef handle(msg): content_type, chat_type, chat_id = telepot.glance(msg) print(content_type, chat_type, chat_id) if content_type == 'text': bot.sendMessage(chat_id, msg['text'])TOKEN = sys.argv[1] # get token from command-linebot = telepot.Bot(TOKEN)MessageLoop(bot, handle).run_as_thread()print ('Listening ...')# Keep the program running.while 1: time.sleep(10)
It is a good habit to always check content_type
before further processing.Do not assume every message is a text
.
Custom Keyboard and Inline Keyboard¶
Besides sending messages back and forth, Bot API allows richer interactionswith custom keyboard andinline keyboard.Both can be specified with the parameter reply_markup
in Bot.sendMessage().The module telepot.namedtuple
provides namedtuple classes for easierconstruction of these keyboards.
Pressing a button on a custom keyboard results in aMessage object sent to the bot,which is no different from a regular chat message composed by typing.
Pressing a button on an inline keyboard results in aCallbackQuery object sentto the bot, which we have to distinguish from a Message object.
Here comes the concept of flavor.
Message has a Flavor¶
Regardless of the type of objects received, telepot generically calls them“message” (with a lowercase “m”). A message’s flavor depends on theunderlying object:
- a Message object gives the flavor
chat
- a CallbackQuery object gives the flavor
callback_query
- there are two more flavors, which you will come to shortly.
Use telepot.flavor() to check a message’s flavor.
Here is a bot which does two things:
- When you send it a message, it gives you an inline keyboard.
- When you press a button on the inline keyboard, it says “Got it”.
Pay attention to these things in the code:
- How I use namedtuple to constructan InlineKeyboardMarkupand an InlineKeyboardButtonobject
- telepot.glance() works on any type of messages. Just give it the flavor.
- Use Bot.answerCallbackQuery() to react to callback query
- To route messages according to flavor, give a routing table toMessageLoop
import sysimport timeimport telepotfrom telepot.loop import MessageLoopfrom telepot.namedtuple import InlineKeyboardMarkup, InlineKeyboardButtondef on_chat_message(msg): content_type, chat_type, chat_id = telepot.glance(msg) keyboard = InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text='Press me', callback_data='press')], ]) bot.sendMessage(chat_id, 'Use inline keyboard', reply_markup=keyboard)def on_callback_query(msg): query_id, from_id, query_data = telepot.glance(msg, flavor='callback_query') print('Callback Query:', query_id, from_id, query_data) bot.answerCallbackQuery(query_id, text='Got it')TOKEN = sys.argv[1] # get token from command-linebot = telepot.Bot(TOKEN)MessageLoop(bot, {'chat': on_chat_message, 'callback_query': on_callback_query}).run_as_thread()print('Listening ...')while 1: time.sleep(10)
Inline Query¶
So far, the bot has been operating in a chat - private, group, or channel.
In a private chat, Alice talks to Bot. Simple enough.
In a group chat, Alice, Bot, and Charlie share the same group. As the humansgossip in the group, Bot hears selected messages (depending on whether inprivacy mode or not) and maychime in once in a while.
Inline query is a totally differentmode of operations.
Imagine this. Alice wants to recommend a restaurant to Zach, but she can’t remember the locationright off her head. Inside the chat screen with Zach, Alice types@Bot where is my favorite restaurant
, issuing an inline query to Bot, likeasking Bot a question. Bot gives back a list of answers; Alice can choose one ofthem - as she taps on an answer, that answer is sent to Zach as a chat message.In this case, Bot never takes part in the conversation. Instead, Bot acts asan assistant, ready to give you talking materials. For every answer Alice chooses,Bot gets notified with a chosen inline result.
To enable a bot to receive InlineQuery,you have to send a /setinline
command to BotFather.An InlineQuery message gives the flavor inline_query
.
To enable a bot to receive ChosenInlineResult,you have to send a /setinlinefeedback
command to BotFather.A ChosenInlineResult message gives the flavor chosen_inline_result
.
In this code sample, pay attention to these things:
- How I use namedtuple InlineQueryResultArticleand InputTextMessageContentto construct an answer to inline query.
- Use Bot.answerInlineQuery() to send back answers
import sysimport timeimport telepotfrom telepot.loop import MessageLoopfrom telepot.namedtuple import InlineQueryResultArticle, InputTextMessageContentdef on_inline_query(msg): query_id, from_id, query_string = telepot.glance(msg, flavor='inline_query') print ('Inline Query:', query_id, from_id, query_string) articles = [InlineQueryResultArticle( id='abc', title='ABC', input_message_content=InputTextMessageContent( message_text='Hello' ) )] bot.answerInlineQuery(query_id, articles)def on_chosen_inline_result(msg): result_id, from_id, query_string = telepot.glance(msg, flavor='chosen_inline_result') print ('Chosen Inline Result:', result_id, from_id, query_string)TOKEN = sys.argv[1] # get token from command-linebot = telepot.Bot(TOKEN)MessageLoop(bot, {'inline_query': on_inline_query, 'chosen_inline_result': on_chosen_inline_result}).run_as_thread()while 1: time.sleep(10)
However, this has a small problem. As you types and pauses,types and pauses, types and pauses … closely bunched inline queries arrive.In fact, a new inline query often arrives before we finish processing a preceding one.With only a single thread of execution, we can only process the closely bunchedinline queries sequentially. Ideally, whenever we see a new inline query coming fromthe same user, it should override and cancel any preceding inline queries being processed(that belong to the same user).
My solution is this. An Answerer takes an inline query, inspects its from
id
(the originating user id), and checks to see whether that user has an unfinished threadprocessing a preceding inline query. If there is, the unfinished thread will be cancelledbefore a new thread is spawned to process the latest inline query. In other words,an Answerer ensures at most one active inline-query-processing thread per user.
Answerer also frees you from having to call Bot.answerInlineQuery() every time.You supply it with a compute function. It takes that function’s returned value and callsBot.answerInlineQuery() to send the results. Being accessible by multiple threads,the compute function must be thread-safe.
import sysimport timeimport telepotfrom telepot.loop import MessageLoopfrom telepot.namedtuple import InlineQueryResultArticle, InputTextMessageContentdef on_inline_query(msg): def compute(): query_id, from_id, query_string = telepot.glance(msg, flavor='inline_query') print('Inline Query:', query_id, from_id, query_string) articles = [InlineQueryResultArticle( id='abc', title=query_string, input_message_content=InputTextMessageContent( message_text=query_string ) )] return articles answerer.answer(msg, compute)def on_chosen_inline_result(msg): result_id, from_id, query_string = telepot.glance(msg, flavor='chosen_inline_result') print ('Chosen Inline Result:', result_id, from_id, query_string)TOKEN = sys.argv[1] # get token from command-linebot = telepot.Bot(TOKEN)answerer = telepot.helper.Answerer(bot)MessageLoop(bot, {'inline_query': on_inline_query, 'chosen_inline_result': on_chosen_inline_result}).run_as_thread()while 1: time.sleep(10)
Maintain Threads of Conversation¶
So far, we have been using a single line of execution to handle messages.That is adequate for simple programs. For more sophisticated programs where states needto be maintained across messages, a better approach is needed.
Consider this scenario. A bot wants to have an intelligent conversation with a lot of users,and if we could only use a single line of execution to handle messages (like what we have done so far),we would have to maintain some state variables about each conversation outside the message-handlingfunction(s). On receiving each message, we first have to check whether the user already has a conversationstarted, and if so, what we have been talking about. To avoid such mundaneness, we need a structured wayto maintain “threads” of conversation.
Let’s look at my solution. Here, I implemented a bot that counts how many messages have been sentby an individual user. If no message is received after 10 seconds, it starts over (timeout).The counting is done per chat - that’s the important point.
import sysimport timeimport telepotfrom telepot.loop import MessageLoopfrom telepot.delegate import pave_event_space, per_chat_id, create_openclass MessageCounter(telepot.helper.ChatHandler): def __init__(self, *args, **kwargs): super(MessageCounter, self).__init__(*args, **kwargs) self._count = 0 def on_chat_message(self, msg): self._count += 1 self.sender.sendMessage(self._count)TOKEN = sys.argv[1] # get token from command-linebot = telepot.DelegatorBot(TOKEN, [ pave_event_space()( per_chat_id(), create_open, MessageCounter, timeout=10),])MessageLoop(bot).run_as_thread()while 1: time.sleep(10)
A DelegatorBot is able to spawn delegates. Above, it is spawning one MessageCounter
per chat id.
Also noteworthy is pave_event_space(). To kill itself after 10 secondsof inactivity, the delegate schedules a timeout event. For events to work, weneed to prepare an event space.
Detailed explanation of the delegation mechanism (e.g. how and when a MessageCounter
is created, and why)is beyond the scope here. Please refer to DelegatorBot.
Inline Handler per User¶
You may also want to answer inline query differently depending on user. When Alice asks Bot“Where is my favorite restaurant?”, Bot should give a different answer than when Charlie asksthe same question.
In the code sample below, pay attention to these things:
- AnswererMixin adds an Answerer instance to the object
- per_inline_from_id() ensures one instance of
QueryCounter
per originating user
import sysimport timeimport telepotfrom telepot.loop import MessageLoopfrom telepot.delegate import pave_event_space, per_inline_from_id, create_openfrom telepot.namedtuple import InlineQueryResultArticle, InputTextMessageContentclass QueryCounter(telepot.helper.InlineUserHandler, telepot.helper.AnswererMixin): def __init__(self, *args, **kwargs): super(QueryCounter, self).__init__(*args, **kwargs) self._count = 0 def on_inline_query(self, msg): def compute(): query_id, from_id, query_string = telepot.glance(msg, flavor='inline_query') print(self.id, ':', 'Inline Query:', query_id, from_id, query_string) self._count += 1 text = '%d. %s' % (self._count, query_string) articles = [InlineQueryResultArticle( id='abc', title=text, input_message_content=InputTextMessageContent( message_text=text ) )] return articles self.answerer.answer(msg, compute) def on_chosen_inline_result(self, msg): result_id, from_id, query_string = telepot.glance(msg, flavor='chosen_inline_result') print(self.id, ':', 'Chosen Inline Result:', result_id, from_id, query_string)TOKEN = sys.argv[1] # get token from command-linebot = telepot.DelegatorBot(TOKEN, [ pave_event_space()( per_inline_from_id(), create_open, QueryCounter, timeout=10),])MessageLoop(bot).run_as_thread()while 1: time.sleep(10)
Async Version (Python 3.5+)¶
Everything discussed so far assumes traditional Python. That is, network operations are blocking;if you want to serve many users at the same time, some kind of threads are usually needed.Another option is to use an asynchronous or event-driven framework, such as Twisted.
Python 3.5+ has its own asyncio
module. Telepot supports that, too.
Here is how to compile and install Python 3.6, if your O/S does not have it built in:
$ sudo apt-get update$ sudo apt-get upgrade$ sudo apt-get install libssl-dev openssl libreadline-dev$ cd ~$ wget https://www.python.org/ftp/python/3.6.1/Python-3.6.1.tgz$ tar zxf Python-3.6.1.tgz$ cd Python-3.6.1$ ./configure$ make$ sudo make install
Finally:
$ pip3.6 install telepot
In case you are not familiar with asynchronous programming, let’s start by learning about generators and coroutines:
… why we want asynchronous programming:
… how generators and coroutines are applied to asynchronous programming:
… and how an asyncio program is generally structured:
Telepot’s async version basically mirrors the traditional version. Main differences are:
- blocking methods are now coroutines, and should be called with
await
- delegation is achieved by tasks, instead of threads
Because of that (and this is true of asynchronous Python in general), a lot of methodswill not work in the interactive Python interpreter like regular functions would.They will have to be driven by an event loop.
Async version is under module telepot.aio
. I duplicate the message counter examplebelow in async style:
- Substitute async version of relevant classes and functions
- Use
async/await
to perform asynchronous operations - Use MessageLoop.run_forever() instead of run_as_thread()
import sysimport asyncioimport telepotfrom telepot.aio.loop import MessageLoopfrom telepot.aio.delegate import pave_event_space, per_chat_id, create_openclass MessageCounter(telepot.aio.helper.ChatHandler): def __init__(self, *args, **kwargs): super(MessageCounter, self).__init__(*args, **kwargs) self._count = 0 async def on_chat_message(self, msg): self._count += 1 await self.sender.sendMessage(self._count)TOKEN = sys.argv[1] # get token from command-linebot = telepot.aio.DelegatorBot(TOKEN, [ pave_event_space()( per_chat_id(), create_open, MessageCounter, timeout=10),])loop = asyncio.get_event_loop()loop.create_task(MessageLoop(bot).run_forever())print('Listening ...')loop.run_forever()