Introduction¶
Amanobot helps you build applications for Telegram Bot API. It works on Python 3.5+ and it also has an async version based 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 amanobot is by reading this page and going through the examples. Let’s go.
Get a token¶
To use the Telegram Bot API, you first have to get a bot account by chatting with BotFather.
BotFather will give you a token, something like 123456789:ABCdefGhIJKlmNoPQRsTUVwxyZ
.
With the token in hand, you can start using amanobot to access the bot account.
Test the account¶
>>> import amanobot
>>> bot = amanobot.Bot('PUT YOUR TOKEN HERE')
>>> bot.getMe()
{'id': 123456789, 'is_bot': True, 'first_name': 'Your Bot', 'username': 'YourBot', 'can_join_groups': True, 'can_read_all_group_messages': False, 'supports_inline_queries': False}
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': 'Alisson',
'id': 200097591,
'last_name': 'L.',
'type': 'private',
'username': 'alissonlauffer'},
'date': 1538346501,
'from': {'first_name': 'Alisson',
'id': 200097591,
'is_bot': False,
'language_code': 'en',
'last_name': 'L.',
'username': 'alissonlauffer'},
'message_id': 149,
'text': 'Hello'},
'update_id': 100000000}]
The chat
field represents the conversation. Its type
can be private
,
group
, supergroup
, or channel
(whose meanings should be obvious, I hope). Above,
Alisson
just sent a private
message to the bot.
According to Bot API, the method getUpdates returns an array of Update objects. As you can see, an Update object is nothing more than a Python dictionary. In amanobot, Bot API objects are represented as dictionary.
Note the update_id
. It is an ever-increasing number. Next time you should use
getUpdates(offset=100000001)
to avoid getting the same old messages over and over.
Giving an offset
essentially acknowledges to the server that you have received
all 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 amanobot
take care of the mundane stuff and notify you whenever new messages arrive:
>>> from amanobot.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 the messages arriving.
Send a message¶
Sooner or later, your bot will want to send you messages. You should have discovered your own user id from above interactions. I will use my real id on this example. Remember to substitute your own id:
>>> bot.sendMessage(200097591, 'Hey!')
Quickly glance
a message¶
When processing a message, a few pieces of information are so central that you
almost always have to extract them. Use amanobot.glance()
to extract
“headline info”. Try this skeleton, a bot which echoes what you said:
import sys
import time
import amanobot
from amanobot.loop import MessageLoop
def handle(msg):
content_type, chat_type, chat_id = amanobot.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-line
bot = amanobot.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 interactions
with custom keyboard and
inline keyboard.
Both can be specified with the parameter reply_markup
in Bot.sendMessage()
.
The module amanobot.namedtuple
provides namedtuple classes for easier
construction of these keyboards.
Pressing a button on a custom keyboard results in a Message 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 a CallbackQuery object sent to 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, amanobot generically calls them “message” (with a lowercase “m”). A message’s flavor depends on the underlying 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 amanobot.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 construct an InlineKeyboardMarkup and an InlineKeyboardButton object
amanobot.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 to
MessageLoop
import sys
import time
import amanobot
from amanobot.loop import MessageLoop
from amanobot.namedtuple import InlineKeyboardMarkup, InlineKeyboardButton
def on_chat_message(msg):
content_type, chat_type, chat_id = amanobot.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 = amanobot.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-line
bot = amanobot.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 humans gossip in the group, Bot hears selected messages (depending on whether in privacy mode or not) and may chime in once in a while.
Inline query is a totally different mode of operations.
Imagine this. Alice wants to recommend a restaurant to Zach, but she can’t remember the location
right off her head. Inside the chat screen with Zach, Alice types
@Bot where is my favorite restaurant
, issuing an inline query to Bot, like
asking Bot a question. Bot gives back a list of answers; Alice can choose one of
them - 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 as
an 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 InlineQueryResultArticle and InputTextMessageContent to construct an answer to inline query.
- Use
Bot.answerInlineQuery()
to send back answers
import sys
import time
import amanobot
from amanobot.loop import MessageLoop
from amanobot.namedtuple import InlineQueryResultArticle, InputTextMessageContent
def on_inline_query(msg):
query_id, from_id, query_string = amanobot.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 = amanobot.glance(msg, flavor='chosen_inline_result')
print ('Chosen Inline Result:', result_id, from_id, query_string)
TOKEN = sys.argv[1] # get token from command-line
bot = amanobot.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 bunched inline queries sequentially. Ideally, whenever we see a new inline query coming from the 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 thread
processing a preceding inline query. If there is, the unfinished thread will be cancelled
before 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 calls
Bot.answerInlineQuery()
to send the results. Being accessible by multiple threads,
the compute function must be thread-safe.
import sys
import time
import amanobot
from amanobot.loop import MessageLoop
from amanobot.namedtuple import InlineQueryResultArticle, InputTextMessageContent
def on_inline_query(msg):
def compute():
query_id, from_id, query_string = amanobot.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 = amanobot.glance(msg, flavor='chosen_inline_result')
print ('Chosen Inline Result:', result_id, from_id, query_string)
TOKEN = sys.argv[1] # get token from command-line
bot = amanobot.Bot(TOKEN)
answerer = amanobot.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 need to 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-handling function(s). On receiving each message, we first have to check whether the user already has a conversation started, and if so, what we have been talking about. To avoid such mundaneness, we need a structured way to maintain “threads” of conversation.
Let’s look at my solution. Here, I implemented a bot that counts how many messages have been sent by 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 sys
import time
import amanobot
from amanobot.loop import MessageLoop
from amanobot.delegate import pave_event_space, per_chat_id, create_open
class MessageCounter(amanobot.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-line
bot = amanobot.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 seconds
of inactivity, the delegate schedules a timeout event. For events to work, we
need 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 asks the same question.
In the code sample below, pay attention to these things:
AnswererMixin
adds anAnswerer
instance to the objectper_inline_from_id()
ensures one instance ofQueryCounter
per originating user
import sys
import time
import amanobot
from amanobot.loop import MessageLoop
from amanobot.delegate import pave_event_space, per_inline_from_id, create_open
from amanobot.namedtuple import InlineQueryResultArticle, InputTextMessageContent
class QueryCounter(amanobot.helper.InlineUserHandler, amanobot.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 = amanobot.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 = amanobot.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-line
bot = amanobot.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¶
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. Amanobot supports that, too.
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:
Amanobot’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 methods will 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 amanobot.aio
. I duplicate the message counter example
below in async style:
- Substitute async version of relevant classes and functions
- Use
async/await
to perform asynchronous operations - Use
MessageLoop.run_forever()
instead ofrun_as_thread()
import sys
import asyncio
import amanobot
from amanobot.aio.loop import MessageLoop
from amanobot.aio.delegate import pave_event_space, per_chat_id, create_open
class MessageCounter(amanobot.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-line
bot = amanobot.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()