Running a Serverless Telegram Bot from AWS Lambda
Function as a service (FaaS) offerings are great for certain compute tasks. They are low maintenance, scalable, and cheap for infrequent workloads.
Telegram bots can be used fairly infrequently, or might be constantly monitoring and moderating your chat with tens of thousands of users. The compute workload is quite variable, so if your chat is dead from 3am to 11am you won’t have to pay to keep a server running 24/7. On the other-hand, during a surge of activity, you won’t need to worry about running out of resources.
In this tutorial, we will use AWS Lambda to run our serverless Telegram bot. You should be able to use any function as a service offering in a similar manner.
Before we begin, check out my previous article to initialize your telegram bot.
I’m assuming you have some knowledge of AWS and Python in this tutorial.
Throughout, you might notice mentions of a “Contact Us Form” — I will try to have both API Gateway and frontend follow-up posts. You can ignore or adjust this data format to your needs, but you will see a bit of code for handling this data along the way.
Our input data will be a simple JSON payload:
{
"name":"Test Name",
"email":"example-email@email.com",
"message":"This is an example message"
}Let’s get this show on the road…
Navigate to Amazon Web Services, log in, and select Lambda. Then create a function:
I’m using Python3.6, but feel free to substitute with your language of choice.
I named my function “send-telegram-message”, and kept the default permissions. With our basic functionality, we won’t need any permissions outside the Lambda Basic Execution Role.
I am a fan of test-driven development for peace-of-mind, and the AWS Lambda console makes it easy. Click on the test button near the top right, then create a test that matches the format of our contact us form.
Since we have not implemented our Lambda function yet, when we run the test it returns a 200 status code and “Welcome from Lambda” in the body.
Let’s start by accessing our incoming data.
If you are not familiar with Lambda, our incoming data is stored in the event object. It is a typical key value mapping, and we can access it with Python’s dictionary syntax. So, let’s grab the data and format it for a Telegram message.
import jsondef lambda_handler(event, context):
name = event['name']
email = event['email']
message = event['message']
telegram_msg =
f'From: {name}\nEmail: {email}\n Message:{message}'
return {
'statusCode': 200,
'body': json.dumps(telegram_msg)
}
But what if a user does not provide a name or email or message? Let’s create another test case called emptyStrings with empty values to simulate an empty form submission:
{
"name": "",
"email": "",
"message": ""
}Well, we run this and it succeeds, but we would receive a meaningless message. How about we provide some feedback by validating the form, and by raising an error if it’s not up to our standards.
We should also provide feedback with our form on the client side to provide a good user experience and to limit unnecessary API calls, but the client side validation can always be bypassed.
Three more test cases for invalid inputs…
emptyTest:
{}whitespaceOnly:
{
“name”: “ “,
“email”: “ “,
“message”: “ “
}and nullEmail:
{
"name": "FirstName LastName",
"email": null,
"message": "Example message!"
}If we can handle all of these cases, we should be pretty good. We might still receive some nonsense input, but at least it’s something we can examine.
Lets initialize a field_errors dictionary and do some form validation. I want all fields to be required, so let’s make a function that ensures the key and value both exist, and then adds an error for the field if there is an issue.
def check_param(event:dict, param:str, field_errors:dict)->str:
# checks for key and if val is null or an empty string
if param in event and event[param] and str(event[param]).strip():
return str(event[param])
else:
field_errors[param] = "Required"
return NoneIf we have any field errors, they will be accumulated and we will raise an exception.
def lambda_handler(event, context):
field_errors = {} # ignore event parameters other than these 3
name = check_param(event, 'name', field_errors)
email = check_param(event, 'email', field_errors)
message = check_param(event, 'message', field_errors)
if field_errors:
raise Exception(json.dumps({'field_errors': field_errors}))
telegram_msg = f'From: {name}\nEmail: {email}\nMessage:{message}'return {
'statusCode': 200,
'body': json.dumps(telegram_msg)
}
Now if you run the above test cases, they should return an expected result.
We have reasonably clean data, so let’s proceed to integrating Telegram into our Lambda function.
If you haven’t already, set up your Telegram bot. In order to send yourself a message, we will need your ChatID and BotToken.
Lamba has a ‘Environment variables’ section right under the ‘Function code’ section of the console, and this is a good spot to add these constant values.
Paste in your respective values without quotes, and we can then access them in the function code. In Python, these values are accessed in the same way as local environment variables. Don’t forget to import the os module!
import os...chat_id = os.environ['MY_CHAT_ID']
telegram_token = os.environ['TELEGRAM_BOT_TOKEN']
Now it is just a matter of sending a request with a valid payload. Add your token to the Telegram API url, and create a parameter payload with your chat_id and formatted message.
from botocore.vendored import requests...api_url = f"https://api.telegram.org/bot{telegram_token}/"
params = {'chat_id': chat_id, 'text': telegram_msg}
res = requests.post(api_url + "sendMessage", data=params).json()
Change your test case back to validContactData, save, and then run the test. You should receive a formatted message on your Telegram client!
Exciting!
Let’s wrap up the Lambda by processing the request response. The Telegram API specifies an “ok” return parameter, so if its valid, we will return a 200 response and the sent message, otherwise we can return a 404 and log the error.
To keep things simple, I will just print the error response to stdout which will be picked up by CloudWatch. Using the python logging module might be a good improvement.
Returning the error has the chance of exposing sensitive information, so I decided to only return the status code.
if res['ok']:
return {
'statusCode': 200,
'body': res['result'],
}
else:
print(res)
return {
'statusCode': 400,
}That wraps up the Lambda function. In summary, my Lambda code looks like this:
import json
import os
from botocore.vendored import requestsdef check_param(event:dict, param:str, field_errors:dict)->str:
# checks for key and if val is null or an empty string
if param in event and event[param] and str(event[param]).strip():
return str(event[param])
else:
field_errors[param] = "Required"
return Nonedef lambda_handler(event, context):
field_errors = {} # ignore event parameters other than these 3
name = check_param(event, 'name', field_errors)
email = check_param(event, 'email', field_errors)
message = check_param(event, 'message', field_errors)
if field_errors:
raise Exception(json.dumps({'field_errors': field_errors}))
telegram_msg = f'From: {name}\nEmail: {email}\nMessage {message}'
chat_id = os.environ['MY_CHAT_ID']
telegram_token = os.environ['TELEGRAM_BOT_TOKEN']
api_url = f"https://api.telegram.org/bot{telegram_token}/"
params = {'chat_id': chat_id, 'text': telegram_msg}
res = requests.post(api_url + "sendMessage", data=params).json()
if res["ok"]:
return {
'statusCode': 200,
'body': res['result'],
}
else:
print(res)
return {
'statusCode': 400,
'body': res
}
Up next, we will expose this Lambda to the web with API Gateway. Check it out here: Integrating Your Serverless Telegram Bot with AWS API Gateway
Sign up for more tutorials & updates at LearnTelegram.com
More sign ups ⇒ more content
Check out my personal site: https://wk0.dev
Thanks for reading.
