A bot for Telegram with Tic-Tac-Toe.
This post is a reissue of the article published in the magazine Portugal a Programar number 55 from March 2017, pages 26 to 34.
In a world with so many instant messaging applications, Telegram stands out for the rich API it provides for creating bots. Bots are small programs that can interact with users and provide services, such as executing commands, managing files or images, and even proposing games!
The Python community has long explored libraries like Telebot and more recently, Telepot. Although the difference in the names of the two is just one letter, the design of Telepot seems more robust to me, and best of all: it integrates asynchronous calls!
The goal of this tutorial is to show how to create an asynchronous bot using Telepot in Python 3.6. It is divided into four parts: why asynchronous? obtaining the key to run the bot, creating the bot, and the game of Tic-Tac-Toe itself (with minimax).
Why asynchronous?#
As a big fan of asynchronous programming, I couldn’t miss the opportunity to write an asynchronous bot. But why asynchronous?
Asynchronous programming came to simplify the lives of programmers who do not want and do not need to use threads. It is extremely useful when the application spends long periods waiting for a response from an input or output device, such as the network. The idea of asynchronous programming is to break the program into tasks. Each task is executed by an event loop, provided by the library responsible for activating and deactivating tasks. When a task needs to execute another asynchronous task, it simply asks the loop to interrupt its own execution and return only when the other task has finished. This cycle of activation and deactivation repeats until the conclusion of our program. Its great advantage is that the processor does not sit idle waiting for a response from the network, a file, or the database. In this way, multiple requests can be handled simultaneously, all without using threads. Although asynchronous programming does not replace parallel programming in all cases, it greatly simplifies the writing of programs that do not need to worry about locking or access concurrency.
In the case of a bot, multiple Telegram users can play at the same time. To improve processor utilization, it is best to write an asynchronous bot that will respond to game commands without blocking one player when another is simply waiting to receive the bot’s response. This is possible because the bot uses REST calls via HTTP to communicate with Telegram. Thus, new commands arrive over the network, and their responses are also sent in HTTP requests, creating a great opportunity to suspend execution during the transmission and reception of commands, freeing the bot to respond to other requests.
In Python 3.5, asynchronous programming saw significant improvements, especially regarding syntax. We can use async def to create a new asynchronous coroutine and await to wait for its execution. We will see more in the bot’s code itself. But to answer the question of this section, asynchronous programming improves processor utilization and responds to multiple players, alternating between them during input/output processing without making them wait. As we now pay for instances on Amazon, Google Compute Engine, or Azure by the hour, executing the maximum number of requests with the minimum number of instances is an important financial pressure for the developer. Asynchronous programming is an alternative to alleviate this pressure without increasing the difficulty of developing the solution itself.
Obtaining the key#
First, if you are not familiar with Telegram, download it at Telegram.org. You can install it on your phone or simply create an account to use on the web. Follow the instructions on the site.
To create a new bot, we must talk to BotFather, a bot manager for Telegram. It is this bot that will create a new bot and an access key that we need to obtain to use the API. You can talk to BotFather by clicking on the link BotFather or searching for @BotFather.
Once you have the chat screen open with BotFather, type the command:
/newbot
Don’t forget to hit Enter at the end of the line.
The creation of the bot is done by “talking” with BotFather. The first thing it asks is for the name of the bot. Choose a new name and type it on a new line. If the name is already in use, BotFather will ask you to choose another.
After accepting the name of your new bot, BotFather displays this message. In the second line, it shows the new link to access your bot on the web. In my case, the bot is called velha_bot. What we are looking for is the key to access the API. In this case, the key for your new bot will be shown after: “Use this token to access the HTTP API.” I painted my key black, but you should have a string with numbers and letters. Copy this line and keep it in a safe place; you will need the key to use the API. The key looks like:
934802948:AODJcjoijcijaoAISDJoiajdoaijAzcWFIS
Great, we just created our bot. Now we need a program to bring it to life!
Creating the bot#
Let’s start by downloading the bot I developed for this article. The source code is available on GitHub. You need to have git and Python 3.6 installed on your machine to continue.
Note: since several Linux distributions still do not have version 3.6 available in their package managers, you can also convert the bot to run on Python 3.5. The differences are minor. Just replace f-strings like f"{name}" with “{}".format(name).
If you are using Windows, install git and Python 3.6 by downloading directly from Python.org.
If you are using Linux, check the installed version on your system by typing:
python -V
Depending on your distribution, you may need to type python3 instead of python. Any version later than 3.6 is fine.
If you want to install another version of the interpreter, follow the instructions below. Ubuntu 18.04 LTS: install git and Python 3.7 with the commands below:
sudo apt update
sudo apt install -y python3.7 wget curl git
and then download the bot with:
git clone https://github.com/lskbr/velha.git
Create a virtualenv using the utility of your choice, such as virtualenvwrapper. Follow the installation instructions if you haven’t installed it yet.
mkvirtualenv velha -p `which python3.7`
Git should download the source code and create a directory called velha on your computer.
Type:
cd velha
pip install -r requirements.txt
Now we will need the key you obtained from BotFather. Replace your key after the =. Type:
export BOT_TOKEN=934802948:AODJcjoijcijaoAISDJoiajdoaijAzcWFIS
and run the bot with:
python velha.py
Great, your bot should now be running!
Let’s test it. Use Telegram’s search feature and look for your bot’s name. In my case, it is @velha_bot, but you should replace it with your bot’s name!
Click or tap START.
The bot responds and asks what difficulty level you want to play. Easy is completely random; the computer will simply fill an empty position. Medium and Hard use a simple but interesting algorithm that we will discuss later. I suggest you choose Hard.
The bot then asks if you want to start playing (X) or if you want to play after the computer (O).
Notice that the previous options have been erased and that the message has been replaced. Choose X.
Finally, we can play. Click on an available position and have fun with Tic-Tac-Toe.
To play again, click Restart. The bot informs the result of the match.
The Game of Tic-Tac-Toe#
Now that we have installed everything, we can see how the program was made! The program turned out to be much larger than I expected, so let’s break it down.
The program starts on line 395, with:
# Get the token from the BOT_TOKEN environment variable
TOKEN = os.getenv("BOT_TOKEN")
if __name__ == "__main__":
logging.basicConfig(level=logging.DEBUG,
format='%(asctime)s %(levelname)6s %(filename)s %(funcName)s:%(lineno)d %(message)s')
loop = asyncio.get_event_loop()
bot = telepot.aio.Bot(TOKEN, loop=loop) # Create the bot
game = TicTacToe(bot) # Create the game
loop.create_task(game.stats(loop)) # Create the task that cleans old games
loop.create_task(bot.message_loop({'chat': game.chat_handler, 'callback_query': game.callback_query}))
try:
loop.run_forever()
except KeyboardInterrupt:
pass
Since the TOKEN should not be stored on GitHub (otherwise it would be public), the program needs to obtain it from the environment. In this case, we use os.getenv to get the token from the BOT_TOKEN variable. That’s why we need to do the export during setup.
After obtaining the TOKEN, we configure the program’s log and the event loop in:
loop = asyncio.get_event_loop()
When we write code for asynchronous Python, we use an event loop. This loop is created and maintained by the asyncio module. Unlike a classic program, an asynchronous program submits routines to be executed in the loop. The loop executes only one routine at a time, which simplifies our code without worrying about parallelism. The great advantage of the loop is that it can stop the execution of a routine while it waits for a device or a network response. We will soon see how Python does this. The downside of the loop is that it only executes one routine at a time, so if a routine takes too long to return execution to the loop, the entire program will be blocked!
Once we have our loop and the token, we create our bot with just one line!
bot = telepot.aio.Bot(TOKEN, loop=loop) # Create the bot
Notice that we use teleport.aio, which is the Telepot module that supports asynchronous programming. We pass the TOKEN and the loop as parameters.
Finally, we create an object of the Game class:
game = TicTacToe(bot) # Create the game
Notice that we pass the bot as a parameter. This is important because once the initialization is complete, it is the loop that will control the execution of our program.
We schedule the execution of a routine that displays how many games we have in memory with:
loop.create_task(game.stats(loop)) # Create the task that cleans old games
Notice that we create the task (create_task) in the loop, passing game.stats(loop). Here things start to differ from a traditional program. The method game.stats is a coroutine and is not executed when called. In fact, it creates a coroutine that can be called by the loop. The code is a bit different from a normal method in Python. The stats method is on line 386:
async def stats(self, loop):
"""Prints statistics and cleans old games"""
while True:
games = len(self.games)
logger.info(f"Games in memory: {games}")
self.games.clean_old()
await asyncio.sleep(60, loop=loop)
Notice that it is defined with async def and not just def! This change allows this method to be called by the loop asynchronously. The rest of the code is practically normal, except for the last line:
await asyncio.sleep(60, loop=loop)
Here we see another novelty, await. This construction makes our asynchronous bot special. When we use await, we ask the loop to stop our current coroutine (created with async def) to execute another coroutine, in this case, asyncio.sleep, which waits 60 seconds to return. Notice that we use asyncio’s sleep. This is important because time.sleep is not a coroutine compatible with the loop, and if we use it, the entire program will stop while it does not return.
With await, we request the execution of asyncio.sleep. The loop then stops the execution of stats, calls asyncio.sleep, and returns to stats when it finishes. The advantage is that asyncio.sleep does not stop the loop, leaving it free to execute other coroutines. And our code remains simple, as if this break in execution had not occurred.
The next step is to configure the bot to call game methods when receiving a message or a response (button clicks):
loop.create_task(bot.message_loop({'chat': game.chat_handler, 'callback_query': game.callback_query}))
Just like we did with game.stats, we will create a coroutine that will be called later by the loop. In this case, we pass a dictionary with the methods we want to call when a chat message is received and for callback_query. In this case, the callback_query is the response that Telegram sends when we click the game buttons.
After that, we let the loop control our program:
try:
loop.run_forever()
except KeyboardInterrupt:
pass
The execution stays within loop.run_forever() until we stop it with Control+C.
The loop waits for events and executes the coroutines we configured. In the case of the bot, it periodically accesses Telegram’s servers to check for new messages. When a message arrives, it calls game.chat_handler:
async def chat_handler(self, msg):
"""Processes the chat coming from the user"""
content_type, chat_type, chat_id = telepot.glance(msg)
logger.debug(f"Content_type: {content_type} Chat type: {chat_type} Message: {msg}")
game = self.get_game(msg)
await self.reply_markup(game, chat_id)
Notice that the chat_handler method is defined with async def, making it a coroutine. It simply receives the message from Telegram, which is a large dictionary with various information about the message, such as the text of the message, who sent it, in which chat, when it was sent, etc.
The Telepot module has some routines that help work with this dictionary, such as glance, which extracts the content type, chat type, and chat_id that we need to respond to a message.
When the bot calls chat_handler, it does not control our game itself, meaning it does not know which user sent the message, but that the bot received a message. What to do with the message is the responsibility of our program. Since we can have multiple games at the same time, we create a dictionary with the games. Each message has a user_id of origin, and this number uniquely identifies each Telegram user (even if they change their name or handle). Thus, using the user_id as a key, we can obtain the game for that user. The Telepot library has some interesting classes that help maintain context by user or by chat; these classes were not used in this example, so consult the documentation for more information.
We then call reply_markup, passing the game and the chat_id of the message so that the bot generates a response:
async def reply_markup(self, game, chat_id=None):
"""Depending on the current state of the game, returns the available options"""
if game.screen == 0:
markup = game.difficulty_level()
message = 'Tic-Tac-Toe - Choose the difficulty level'
elif game.screen == 1:
markup = game.player_type()
message = 'X always plays first. Do you want to play as X or O?'
elif game.screen == 2:
markup = game.build_grid()
message = game.message or 'Tic-Tac-Toe'
if game.message is None and chat_id is not None:
message = await self.bot.sendMessage(chat_id, message, reply_markup=markup)
game.message = telepot.message_identifier(message)
else:
try:
await self.bot.editMessageText(game.message, message, reply_markup=markup)
except telepot.exception.TelegramError as te:
pass
You may have noticed that the bot asks for the difficulty level, whether we want to play with X or O before letting us play. Each of these steps is controlled by a property of each game that keeps track of where we are for each player. When screen is 0, we present the options for choosing the difficulty level. When it is 1, we ask if the player wants to play with X or O, 2 is the game itself, and 3 is a message informing that the match has already ended.
Notice that game.message is a variable we use to store the message identifier. This is necessary because to avoid generating a new grid with each response, we edit the previous message (the previous one is replaced by a new one, giving the impression of disappearing). Thus, if game.message is null, we send a new message with self.bot.sendMessage. Notice that we use await because sendMessage is a coroutine of the bot. This line is a bit special because we are using await and assigning its result to the message variable. This is another great advantage of await; it returns the same values from the coroutine that just executed. In this case, during the time it takes to send this message, the loop remains free to perform other tasks. When sendMessage returns, the program resumes execution at the line of await, and message receives its return. Notice that we immediately store the message identifier in game.message. This way, the next time, we will edit the previous message instead of sending a new one.
In the case of editMessageText, we replace the content of the previous message with our new screen. A basic exception handling is performed because a user may click on an old game and generate errors (not handled here to simplify the bot).
Each screen of the game consists of a text part and a series of buttons, specified in the reply_markup parameter of sendMessage and editMessageText. The markup we send is generated based on the current state of the game. Let’s see where we create the markup with the difficulty level:
def difficulty_level(self):
"""Returns the options for the difficulty levels of the game"""
return InlineKeyboardMarkup(
inline_keyboard=[
[InlineKeyboardButton(text="Easy", callback_data='easy')],
[InlineKeyboardButton(text="Medium", callback_data='medium')],
[InlineKeyboardButton(text="Hard", callback_data='hard')]])
We are now in the code of the TicTacToe class, which keeps the state of the game for each player (line 152). The method difficulty_level is defined on line 195. In this case, we simply pass a list of InlineKeyboardButton, each with its text and callback_data. The text is shown to the user inside the button, and the callback_data is sent by Telegram if this button is pressed.
This is how we know which option was chosen by the user. When a response from the button is received, the bot calls our callback_query method (line 357):
async def callback_query(self, msg):
"""Processes the response to the user's choices"""
query_id, from_id, query_data = telepot.glance(msg, flavor='callback_query')
logger.debug(f'Callback Query: {query_id}, {from_id}, {query_data}')
game = self.get_game(msg)
logger.debug(f"Callback query: user: {game.user_id} message: {msg}")
if game.screen == 0:
self.set_difficulty(game, query_data)
await self.reply_markup(game, self.msg_chat_id(msg))
elif game.screen == 1:
self.set_first_player(game, query_data)
if game.computer == "X":
self.play_for_computer(game)
await self.reply_markup(game, self.msg_chat_id(msg))
elif query_data == "restart":
game = self.games.new_game(game)
await self.reply_markup(game, self.msg_chat_id(msg))
elif len(query_data) == 1 and query_data.isdigit() and game.screen == 2:
position = int(query_data) - 1
if game.play(position, game.player):
self.check_move(game)
grid = game.build_grid()
await self.bot.editMessageText(game.message, f"Tic-Tac-Toe: {game.message}", reply_markup=grid)
else:
await self.bot.answerCallbackQuery(query_id, text='Choose another position')
elif game.screen == 3:
await self.bot.answerCallbackQuery(query_id, text='Match finished. Choose restart to play again')
Just like when we receive a text message, a message response from the buttons is received. The response itself comes in query_data, extracted by telebot.glance (notice that we pass a flavor parameter=‘callback_query’):
query_id, from_id, query_data = telepot.glance(msg, flavor='callback_query')
The query_data has the same text we specified in the callback_data of our buttons. The rest is traditional Python, where we check the state of the game and execute the necessary configurations.
When we are on screen 2 and receive a number, it means that the user clicked on the grid. The grid of Tic-Tac-Toe is organized into 9 positions, numbered from 1 to 9. Since Python indexes from zero, we convert this response into an integer and subtract 1 to get the respective index or the position they played.
If you observe the source code, you will see that the state of each game is stored in a list of strings with nine elements. Each element represents a position on the grid. A blank space for empty positions, X or O if one of the players has already marked that position previously.
When we call game.play(position, game.player), we ask for a new mark to be made. This method returns False if the user has already marked this position, and the code generates an error message:
await self.bot.answerCallbackQuery(query_id, text='Choose another position')
Notice that answerCallbackQuery shows only an overlay in Telegram and does not alter our message itself. Try to play twice in the same position to see the effect. We will use this method to send important messages to the user, especially error messages.
In game.check_move, we test if the game is over and return the new grid.
This is the general mechanism of the game. Let’s take a look at how we made the computer play.
And the computer plays Tic-Tac-Toe#
The game of Tic-Tac-Toe is quite simple, and it would be very boring to have to look for another player for a match. Let’s see how to make the computer play. Using a technique called Minimax Minimax. An excellent explanation of this algorithm for solving Tic-Tac-Toe can be found in Portuguese here: http://www.maawko.com/blog/freecodecamp/jogo-da-velha-entendendo-o-algoritimo-minimax/
The original article in English can be read here: http://neverstopbuilding.com/minimax.
Unfortunately, the author wrote the solution using Ruby (just kidding) and not in Python.
The method consists of generating the possible moves for each player, based on the still free positions. After that, we check the result if we played in that position; did we win? Did we lose? Or did we draw? The general idea is to assign a certain number of points when we win and the inverse of this number for the opposing player.
This work is done by the Velhus class, defined on Line 24:
class Velhus:
"""
Class that simulates the grid and allows calculating possible moves.
Used to calculate the computer's move.
The State contains the grid as a list of strings.
A space means that the position is free.
X or O indicates that the player has already marked this position.
Grid
Indices Positions
0 1 2 1 | 2 | 3
---+---+---
3 4 5 4 | 5 | 6
---+---+---
6 7 8 7 | 8 | 9
"""
WINNING = [set(x) for x in [(0, 1, 2), (3, 4, 5), (6, 7, 8),
(0, 4, 8), (6, 4, 2),
(0, 3, 6), (1, 4, 7), (2, 5, 8)]]
def __init__(self, state=None):
"""state: initial state. Default: empty"""
self.state = state or [" "] * 9
def possible_moves(self):
"""Where can we play?"""
return positions_of(self.state, " ")
def positions_by_player(self):
"""Returns a tuple with the positions of player X and player O"""
return (positions_of(self.state, "X"), positions_of(self.state, "O"))
def won(self, positions, player):
"""Checks if one of the players has won the match"""
s = set(positions)
for p in Velhus.WINNING:
if len(p - s) == 0:
return True
return False
def play(self, position, player):
"""Plays for the player in a specific position"""
if self.state[position] == " ":
self.state[position] = player
else:
raise ValueError(f"Position({position}) invalid.")
def result(self):
jX, jO = self.positions_by_player()
if self.won(jX, "X"):
return("X") # X Won
elif self.won(jO, "O"):
return("O") # O Won
elif not self.possible_moves():
return("*") # Draw without result
else:
return("?") # Inconclusive
@staticmethod
def switch(player):
"""Switches the current player. X --> O and O --> X"""
return "X" if player == "O" else "O"
@staticmethod
def best(result, player):
if player == "X":
return max(result.values())
else:
return min(result.values())
def next(self, player, state, level=0, nmax=3):
"""Creates a dictionary that calculates the possibilities of future moves.
player: current player (whose turn it is)
state: state of the game (grid)
level: current level of recursion, used to decrease the difficulty of the game
nmax: maximum level of exploration. Returns if the current level reaches the maximum.
result: dictionary with the score by result.
"""
result = {}
# Iterates through the possible moves
for possible in self.possible_moves():
j = Velhus(state[:]) # Creates a hypothetical board from the current state.
j.play(possible, player) # plays for the player
outcome = j.result() # checks the result of the move
if outcome == "X" or outcome == "O":
rlocal = 10 - level # Assigns points based on the current level
result[possible] = rlocal if outcome == "X" else -rlocal
elif outcome == "?" and level < nmax: # Since the result is not final, continues to play
other = self.switch(player)
lresult = j.next(other, j.state, level + 1, nmax)
result[possible] = self.best(lresult, other) if lresult else 0
return result
def best_move(self, player, state, dmax):
"""
Calculates the best move for the player
player: the player whose best move will be calculated
state: the current state of the game
dmax: maximum depth level. Used to decrease difficulty.
"""
result = self.next(player, state, nmax=dmax) # What are the possibilities?
best_moves = []
best = self.best(result, player)
logger.debug(Velhus.dump_state(state))
logger.debug(f"Player={player}")
for position, outcome in result.items():
if outcome == best: # If this position has the best score
best_moves.append(position)
logger.debug(f"Best: {best} {best_moves} r={outcome} position={position}")
return best_moves
Some methods were removed to save space and facilitate understanding. Let’s analyze the next method. This method receives the player and the state of the grid.
Then, for each still free position on the grid:
for possible in self.possible_moves():
We create a new instance of Velhus and play in each of these positions, one by one. Notice that the state of Velhus is independent of the state of the game, as we are still checking where to play, meaning we do all this without altering the state of the game itself.
j = Velhus(state[:]) # Creates a hypothetical board
j.play(possible, player) # plays for the player
outcome = j.result() # checks the result of the move
Each move has a result that can be a victory for X, a victory for O, a draw, or indeterminate (meaning that no one has won or lost yet).
With the result in hand, we proceed to the exception of minimax:
if outcome == "X" or outcome == "O":
rlocal = 10 - level # Assigns points based on the current level
result[possible] = rlocal if outcome == "X" else -rlocal
elif outcome == "?" and level < nmax: # Since the result is not final, continues to play
other = self.switch(player)
lresult = j.next(other, j.state, level + 1, nmax)
result[possible] = self.best(lresult, other) if lresult else 0
As dictated by the algorithm, we will assign 10 points for a victory (X indicates victory for player X and O for their opponent). From these points, we will deduct the level to penalize results that occur after multiple moves. An important detail is that the result for player X is positive and for O negative, hence the sign inversion!
If the result is indeterminate, we recursively call next, but switch the player to simulate their moves. Notice that we pass the new state (already with our hypothetical move) and increment the level. We use the level to limit the expansion of the game and create the medium difficulty level (which does not evaluate all possibilities). Then we evaluate the result of next, but from the perspective of the opposing player. Notice that the best method is crucial because it decides whether we are maximizing or minimizing the result:
@staticmethod
def best(result, player):
if player == "X":
return max(result.values())
else:
return min(result.values())
As dictated by the Minimax algorithm, player X maximizes their moves, and player O minimizes them!
The next method simply calculates the values for all possible moves. Thus, the best_move method can decide where the computer will play:
def best_move(self, player, state, dmax):
result = self.next(player, state, nmax=dmax) # What are the possibilities?
best_moves = []
best = self.best(result, player)
for position, outcome in result.items():
if outcome == best: # If this position has the best score
best_moves.append(position)
return best_moves
We apply the best method again to filter the best move. The difference is that the best_move method supports multiple positions with the same result, meaning multiple positions where the result is equal to the maximum. The best moves are returned, and the computer simply chooses one of them at random.
This cycle of playing, marking, evaluating results repeats until the end of the match.
Conclusion#
I hope you enjoyed this solution and have fun creating other bots for Telegram. Each bot is unique, and we have only scratched the surface of what can be done with Telepot. Telegram also provides all the documentation for its API on its site.
If you do not program in Python, I invite you to get to know this language and its community. Asynchronous programming in Python is excellent, and the number of libraries grows every day. Databases, websites, networks, everything can now be accessed asynchronously.
Writing bots is fun; you can see another bot I wrote using the Telebot library on GitHub: https://github.com/PyNorte/pybrtelegrambot. This bot uses an SQLite database to respond to commands and store data from members of the Python Group of Northern Brazil. Although it is not an asynchronous bot, I am sure it will spark your curiosity to write bots!
About Nilo Menezes#
Nilo Ney Coutinho Menezes is a software developer, specialized in parallel, asynchronous programming, and distributed systems. He has worked on various European projects as a researcher in the areas of simulation, mobile telephony, and networks. He coordinated software development teams for industries in Manaus, Brazil. Today he works in his company in Belgium, providing consulting for the development of scalable systems and cloud computing. He holds a master’s degree in computer science and a bachelor’s degree in data processing from the Federal University of Amazonas. He is the author of the book “Introduction to Programming with Python,” published by Editora Novatec and available in both Brazil and Portugal. Nilo can be found at: https://python.nilo.pro.br or on Telegram @lskbr and https://t.me/PyCoding.