This article is written by Alparslan Mesri and Saki Mesri.
1. IntroductionđȘ
I always liked the idea of asking Professor Dumbledore for advice. He is calm, wise, and like the perfect person to ask when you have a problem. So, one day I thought: what if I make a Telegram bot that talks like Dumbledore?
The idea was simple. When I send a normal message, the bot should just send it back to me. But when I ask something that sounds like a âDumbledore question,â then the bot should answer in a Dumbledore-style voice. Not exactly magic, but close enough.
To make this work, I built a system on AWS. Each part has a different job, and the whole flow feels pretty clean. I also store messages in DynamoDB. Iâll explain the architecture in the next section.
In this article, I will explain how the pieces connect and how I deployed everything using AWS SAM. Itâs not a huge project, but I learned a lot from it. Maybe someone else who wants to build a chatbot â or just wants Dumbledore to talk on their phone â might find this useful.
2. Project Architecture OverviewđȘ
The architecture looks a little long when you list all the services, but the idea is simple:
Telegram sends a message â AWS processes it â the bot sends something back.
Here is the basic flow:
- Telegram sends the userâs message to my API Gateway endpoint.
- API Gateway triggers a Lambda function, which puts the text into SQS.
- Another Lambda takes the message from SQS and calls Amazon Lex.
- Lex decides if this message is a âDumbledore questionâ or not.
- If yes, a third Lambda sends the text to the ChatGPT API and then responds on Telegram.
- If not, the bot just replies with the same message the user sent.
- Every message is saved into DynamoDB, mostly for debugging and curiosity.
So it sounds like many services, but each one is doing a very small job:
- API Gateway â receives requests from Telegram
- Lambda â small functions that connect everything
- SQS â keeps the system safe and prevents overload
- Amazon Lex â decides if this is a Dumbledore question
- OpenAI ChatGPT API â writes the magical answer
- DynamoDB â logs every message
- AWS SAM â used for deployment
This way, the bot stays simple and serverless, and I donât have to run any servers by myself.
In the next section, I will explain what the bot actually does in more detail.
3. What the Bot Actually DoesđȘ
From the userâs point of view, the bot is very simple. You send a message on Telegram, and it replies. But inside, the bot tries to guess if you are talking to Dumbledore or just sending a normal message.
For example:
- If I send: âHelloâ
The bot replies: âYou said: Helloâ - If I send: âDumbledore, what should I do when I feel lost?â
Then the bot asks ChatGPT to write something wise and a bit magical, like Dumbledore would say.
Most of the time, this decision is done by Amazon Lex. I created a small intent that tries to detect messages that sound like a question for Dumbledore. If Lex finds the intent, then the message goes to the Lambda that calls the ChatGPT API. If not, the message still goes to the same Lambda. In that case, the bot simply replies with something like âYou said: Hello.â
So the logic is:
- User sends message
- Lex tries to understand it
- If it feels like a Dumbledore question â call ChatGPT and answer like Dumbledore
- Otherwise â send the same text back
Itâs a very simple rule, but even this small idea makes the bot feel a bit magical.
Also, itâs funny when the bot answers a deep, poetic message, even if the user was expecting something boring.
4. Step-by-Step ImplementationđȘ
Below I walk through how I built each piece. Iâll show the code snippets needed.
1) Create the Telegram bot & webhook
- Talk to @BotFather on Telegram and create a bot. Save the bot token (
TELEGRAM_BOT_TOKEN). - Create an HTTPS endpoint with API Gateway (weâll deploy with SAM later). After deploy youâll have an endpoint like
https://.../telegram. - Set the webhook:
curl -X POST "https://api.telegram.org/bot<TELEGRAM_BOT_TOKEN>/setWebhook" \
-d "url=https://<API_GATEWAY_URL>/telegram"
ă»Telegram will POST update objects to your endpoint when users message the bot.
2) Lambda 1 â Receive Telegram updates & push to SQS
Purpose: receive the webhook, extract text + user info, write to DynamoDB (log), then push to SQS for async processing.
Environment variables
SQS_QUEUE_URLUSERS_TABLE_NAME
Python
import boto3
import os
import json
import urllib.request
from datetime import datetime
sqs = boto3.client('sqs')
queue_url = os.environ['SQS_QUEUE_URL']
# DynamoDB
dynamodb = boto3.resource('dynamodb')
users_table = dynamodb.Table(os.environ['USERS_TABLE_NAME'])
def lambda_handler(event, context):
print("Raw event:", event)
body = event.get("body")
if not body:
print("No body found in event")
return {"statusCode": 200, "body": "No body received"}
try:
telegram_update = json.loads(body)
except Exception as e:
print("JSON decode error:", e)
return {"statusCode": 200, "body": "Invalid JSON"}
print("Telegram update:", telegram_update)
chat_id = telegram_update.get("message", {}).get("chat", {}).get("id")
message_text = telegram_update.get("message", {}).get("text", "")
print("Message text:", message_text)
# ------------------------------
# DynamoDB Users table operations
# ------------------------------
try:
response = users_table.get_item(Key={"user_id": str(chat_id)})
user_exists = 'Item' in response
now_iso = datetime.utcnow().isoformat()
if not user_exists:
users_table.put_item(
Item={
"user_id": str(chat_id),
"last_seen": now_iso
}
)
print(f"Created new user: {chat_id}")
else:
users_table.update_item(
Key={"user_id": str(chat_id)},
UpdateExpression="SET last_seen = :t",
ExpressionAttributeValues={":t": now_iso}
)
print(f"Updated last_seen for user: {chat_id}")
except Exception as e:
print("Error updating Users table:", e)
# ------------------------------
# Send message to SQS
# ------------------------------
try:
response = sqs.send_message(
QueueUrl=queue_url,
MessageBody=json.dumps({
"chat_id": chat_id,
"text": message_text
})
)
print("SQS response:", response)
except Exception as e:
print("Error while sending to SQS:", e)
# Notify Telegram only if SQS send fails
token = os.environ["TELEGRAM_TOKEN"]
url = f"https://api.telegram.org/bot{token}/sendMessage"
data = {"chat_id": chat_id, "text": "â ïž Your message could not be processed. Please try again later."}
try:
req = urllib.request.Request(
url,
data=json.dumps(data).encode(),
headers={"Content-Type": "application/json"}
)
urllib.request.urlopen(req)
except Exception as e2:
print("Failed to send Telegram reply:", e2)
return {"statusCode": 200, "body": "Processed"}
3) SQS â buffer messages
Why SQS?
- Makes the system resilient (Telegram -> web request wonât fail if downstream is slow)
- Allows rate-limiting and retry policies
Queue config suggestions:
- Standard queue is fine.
- Set redrive policy to a DLQ for failed items.
4) Lambda 2 â consumer: call Lex
Lambda 2 reads messages from SQS, calls Amazon Lex to detect the intent, sends a reply back to Telegram, and logs everything in DynamoDB. Hereâs how it works:
Environment variables
LEX_BOT_ALIAS_IDLEX_BOT_IDLEX_LOCAL_IDLOGS_TABLE_NAMETELEGRAM_TOKEN
Python
import json
import boto3
import os
import urllib.request
from datetime import datetime, timezone, timedelta
import time
lex_client = boto3.client("lexv2-runtime")
dynamodb = boto3.resource("dynamodb")
logs_table = dynamodb.Table(os.environ["LOGS_TABLE_NAME"]) # Logs table name from environment variable
BOT_ID = os.environ.get("LEX_BOT_ID")
BOT_ALIAS_ID = os.environ.get("LEX_BOT_ALIAS_ID")
BOT_LOCALE_ID = os.environ.get("LEX_LOCALE_ID", "en_US")
TELEGRAM_TOKEN = os.environ.get("TELEGRAM_TOKEN") # Telegram Bot Token
def lambda_handler(event, context):
for record in event.get("Records", []):
print(f"SQS event: {record}")
body_str = record.get("body", "")
try:
body = json.loads(body_str)
except json.JSONDecodeError:
print(f"Invalid JSON in SQS body: {body_str}")
continue
user_id = str(body.get("chat_id", "anonymous"))
text = body.get("text", "")
print(f"Received from SQS: {text}")
# ----------------------------
# Send text to Amazon Lex
# ----------------------------
reply_text = ""
try:
lex_response = lex_client.recognize_text(
botId=BOT_ID,
botAliasId=BOT_ALIAS_ID,
localeId=BOT_LOCALE_ID,
sessionId=user_id,
text=text
)
print("Lex response:", json.dumps(lex_response))
if "messages" in lex_response:
reply_text = " ".join([m.get("content", "") for m in lex_response["messages"]])
if reply_text:
# Send reply back to Telegram
url = f"https://api.telegram.org/bot{TELEGRAM_TOKEN}/sendMessage"
data = {"chat_id": user_id, "text": reply_text}
req = urllib.request.Request(
url,
data=json.dumps(data).encode("utf-8"),
headers={"Content-Type": "application/json"}
)
urllib.request.urlopen(req)
print(f"Sent to Telegram: {reply_text}")
except Exception as e:
print("Error calling Lex or Telegram:", str(e))
reply_text = "" # Even if failed, continue logging
# ----------------------------
# Save logs to DynamoDB
# ----------------------------
ttl_seconds = 60 * 60 # TTL = 60 minutes
expire_at = int(time.time()) + ttl_seconds
# UNIX timestamp (for numeric sorting or machine usage)
unix_ts = int(time.time())
# Human-readable timestamp (ISO8601)
readable_ts = datetime.now(timezone(timedelta(hours=9))).strftime("%Y-%m-%d %H:%M:%S")
try:
timestamp = datetime.utcnow().isoformat()
logs_table.put_item(
Item={
"user_id": user_id,
"timestamp": unix_ts,
"timestamp_str": readable_ts,
"input_message": text,
"bot_response": reply_text,
"expire_at": expire_at # TTL attribute
}
)
print(f"Saved log for user {user_id} at {timestamp}")
except Exception as e:
print("Error saving to Logs table:", str(e))
return {"statusCode": 200}
Notes
- This Lambda handles all messages, whether they are Dumbledore questions or not. Lex decides the intent; if it detects a Dumbledore question,
reply_textwill be a magical answer. Otherwise, the Lambda can just reply with something likeYou said: .... - Logging: All messages and bot responses are saved in DynamoDB with a TTL, so logs automatically expire after 1 hour. (We can change expired time.)
- Lesson from experience (SQS retry hell!)
When I first set this up, I forgot to configure a Dead Letter Queue (DLQ) for SQS. One day, my Lambda had a minor bug and failed to process a message. SQS kept retrying⊠over and over.
The same bad message kept showing up in CloudWatch logs every few seconds. It felt like the bot was possessed. After I set up a DLQ, any failed message automatically went there after the retry limit, and I could inspect it manually. No more infinite retries.
Lesson learned: always configure a DLQ â it saves your sanity. - Robustness: When Lambda processes multiple SQS messages, some messages might fail â for example, a call to Lex or Telegram could timeout. I made sure the Lambda catches exceptions per message so that one bad message doesnât stop all the others. This way, the bot keeps running smoothly, and failed messages can be inspected later (or sent to a DLQ).
5) Lex
- Amazon Lex is responsible for detecting whether a message is a Dumbledore-style question. I created a main intent called
AskProfessorIntentthat tries to match questions like:
- But sometimes users write something that doesnât sound like a Dumbledore question. For those cases, Lex needs a fallback mechanism, so I created a
FallbackIntentas well. - The Fulfillment Lambda code hook is turned on for both
AskProfessorIntentandFallbackIntent, allowing Lambda 3 to be called no matter which intent Lex identifies.
6) Lambda 3 â generate response & send back to Telegram
Purpose: if Lex says DumbledoreIntent, call OpenAI ChatGPT API to generate a Dumbledore-ish message. Otherwise return an echo-style response.
Environment variables
OPENAI_API_KEY
Python
import json
import os
import urllib.request
OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY")
def call_chatgpt(user_input: str) -> str:
"""
Sample function to call the ChatGPT API (using urllib)
"""
print("### ChatGPT request:", user_input)
url = "https://api.openai.com/v1/chat/completions"
headers = {
"Authorization": f"Bearer {OPENAI_API_KEY}",
"Content-Type": "application/json"
}
payload = {
"model": "gpt-4o-mini",
"messages": [
{"role": "system", "content": "You are Dumbledore in Harry Potter. Answer like him"},
{"role": "user", "content": user_input}
]
}
req = urllib.request.Request(
url,
data=json.dumps(payload).encode("utf-8"),
headers=headers,
method="POST"
)
try:
with urllib.request.urlopen(req) as response:
resp_data = json.loads(response.read().decode("utf-8"))
# Check if successful response
if "choices" in resp_data:
return resp_data["choices"][0]["message"]["content"]
elif "error" in resp_data:
return f"OpenAI API error: {resp_data['error'].get('message', 'Unknown error')}"
else:
return "Unexpected response format from OpenAI API."
except Exception as e:
print("Error calling OpenAI:", str(e))
return f"Sorry, I could not get a response from ChatGPT. Error: {str(e)}"
def echo_message(user_input: str) -> str:
#Return user input as-is
return f"You said: {user_input}"
def lambda_handler(event, context):
print("Incoming event:", json.dumps(event))
intent = event["sessionState"]["intent"]["name"]
user_input = event.get("inputTranscript", "")
print("### Identified intent:", intent)
print("### User input:", user_input)
if intent == "AskProfessor":
print("### Calling ChatGPT API ###")
answer = call_chatgpt(user_input)
elif intent == "FallbackIntent":
print("### Fallback triggered (echo) ###")
answer = echo_message(user_input)
else:
print("### Unknown intent:", intent)
answer = "Sorry, I couldn't understand that.hahaha"
print("### Final answer to Lex:", answer)
# Response format for Lex V2
return {
"sessionState": {
"dialogAction": {"type": "Close"},
"intent": {"name": intent, "state": "Fulfilled"}
},
"messages": [
{"contentType": "PlainText", "content": answer}
]
}
7) DynamoDB logging pattern
I created two DynamoDB tables: Users and Logs.
- The
Userstable hasuser_idas its partition key. - The
Logstable hasuser_idas the partition key andtimestampas the sort key.
Benefits:
- Easy to query conversation history for a chat.
- Helps when testing intents / tuning Lex utterances.
8) Deploy with AWS SAM
Because of dependencies, I created two template.yaml files. First, I deploy Lex-stack to create the Lex bot. Then, using the information from the deployed Lex bot, I deploy Backend-stack, which creates all the other services.
Before deploying, make sure your local environment meets the following requirements:
- AWS CLI installed and configured
- Youâll need valid AWS credentials (
aws configure)
2. AWS SAM CLI installed
- Required to build and deploy Lambda, API Gateway, SQS, etc. from the template
3. Python installed locally
- The local Python version must match the Lambda runtime defined in
template.yaml
(e.g., if your Lambda runtime ispython3.11, install Python 3.11 locally) - SAM uses your local Python interpreter to build dependencies inside the deployment package
Lex-stack template.yaml:
AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: Lex
Globals:
Function:
Runtime: python3.11
Timeout: 15
MemorySize: 128
Tracing: Active
Resources:
################################
# IAM Role for Lex
################################
LexRole:
Type: AWS::IAM::Role
Properties:
AssumeRolePolicyDocument:
Version: '2012-10-17'
Statement:
- Effect: Allow
Principal:
Service: lexv2.amazonaws.com
Action: sts:AssumeRole
Policies:
- PolicyName: LexPolicy
PolicyDocument:
Version: '2012-10-17'
Statement:
# For Lambda
- Effect: Allow
Action:
- lambda:InvokeFunction
- lambda:GetFunction
Resource: !Sub "arn:aws:lambda:${AWS::Region}:${AWS::AccountId}:function:Sam-TeleBot-dev-a-lambda-LexIntentHandler"
# For CloudWatch Logs
- Effect: Allow
Action:
- logs:CreateLogGroup
- logs:CreateLogStream
- logs:PutLogEvents
Resource: !Sub "arn:aws:logs:${AWS::Region}:${AWS::AccountId}:log-group:/aws/lex/*"
################################
# Lex Bot (CloudFormation)
################################
MyLexBot:
Type: AWS::Lex::Bot
Properties:
RoleArn: !GetAtt LexRole.Arn
DataPrivacy:
ChildDirected: false
IdleSessionTTLInSeconds: 300 # 5min
Name: !Sub "${AWS::StackName}-lex"
BotLocales:
- LocaleId: "en_US"
NluConfidenceThreshold: 0.40
Intents:
- Name: "FallbackIntent"
Description: "Default intent when no other intent matches. The chatbot will echo the message."
ParentIntentSignature: "AMAZON.FallbackIntent"
FulfillmentCodeHook:
Enabled: true
- Name: "AskProfessor"
Description: "This intent is used when the user wants to ask a question to Professor Dumbledore. When the user types commands like ask Dumbledore, the bot will forward the question to ChatGPT and return an answer in the style of Dumbledore."
SampleUtterances:
- Utterance: "ask the professor"
- Utterance: "Dumbledore please answer my question"
- Utterance: "I want to ask Dumbledore something"
- Utterance: "ask Dumbledore about magic"
FulfillmentCodeHook:
Enabled: true
#TargetLambdaArn: !Sub "arn:aws:lambda:${AWS::Region}:${AWS::AccountId}:function:Sam-TeleBot-dev-a-lambda-LexIntentHandler"
################################
# Lex Bot Version / Alias
################################
MyLexBotVersion:
Type: AWS::Lex::BotVersion
DependsOn: MyLexBot
Properties:
BotId: !Ref MyLexBot
BotVersionLocaleSpecification:
- LocaleId: en_US
BotVersionLocaleDetails:
SourceBotVersion: DRAFT
MyLexBotAlias:
Type: AWS::Lex::BotAlias
DependsOn: MyLexBotVersion
Properties:
BotId: !Ref MyLexBot
BotVersion: !GetAtt MyLexBotVersion.BotVersion
BotAliasName: !Sub "${AWS::StackName}Prod"
SentimentAnalysisSettings:
DetectSentiment: false
################################
# Outputs (export for Backend-stack)
################################
Outputs:
LexBotId:
Description: Lex Bot ID
Value: !Ref MyLexBot
Export:
Name: LexBotId-Export
LexBotAliasId:
Description: Lex Bot Alias ID
Value: !GetAtt MyLexBotAlias.BotAliasId
Export:
Name: LexBotAliasId-Export
Backend-stack template.yaml:
AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: Telegram Webhook â API Gateway â Lambda1 â SQS(process/DLQ) â Lambda2 â Lex â Lambda3 +DynamoDB
Globals:
Function:
Runtime: python3.11
Timeout: 15
MemorySize: 128
Tracing: Active
Resources:
################################
# DynamoDB
################################
UsersTable:
Type: AWS::Serverless::SimpleTable
Properties:
PrimaryKey:
Name: user_id
Type: String
TableName: !Sub "${AWS::StackName}-Users"
LogsTable:
Type: AWS::DynamoDB::Table
Properties:
TableName: !Sub "${AWS::StackName}-Logs"
AttributeDefinitions:
- AttributeName: user_id
AttributeType: S
- AttributeName: timestamp
AttributeType: N
KeySchema:
- AttributeName: user_id
KeyType: HASH # Partition key
- AttributeName: timestamp
KeyType: RANGE # Sort key
BillingMode: PAY_PER_REQUEST
################################
# SQS Queue (DLQ)
################################
DLQ:
Type: AWS::SQS::Queue
Properties:
QueueName: !Sub "${AWS::StackName}-sqs-dlq"
MessageRetentionPeriod: 1209600 # 14days
VisibilityTimeout: 60
################################
# SQS Queue (process)
################################
ProcessingQueue:
Type: AWS::SQS::Queue
Properties:
QueueName: !Sub "${AWS::StackName}-sqs"
MessageRetentionPeriod: 1209600 # 14days
VisibilityTimeout: 60
RedrivePolicy:
deadLetterTargetArn: !GetAtt DLQ.Arn
maxReceiveCount: 3
################################
# API Gateway
################################
TelegramWebhookApi:
Type: AWS::Serverless::HttpApi
Properties:
Name: !Sub "${AWS::StackName}-apigw"
StageName: dev
################################
# Lambda1 (Telegram â SQS + DynamoDB)
################################
Lambda1:
Type: AWS::Serverless::Function
Properties:
FunctionName: !Sub "${AWS::StackName}-lambda-TelegramToSQS"
CodeUri: lambda1/
Handler: lambda_function.lambda_handler
Runtime: python3.11
Environment:
Variables:
USERS_TABLE_NAME: !Ref UsersTable
SQS_QUEUE_URL: !Ref ProcessingQueue
Policies:
- DynamoDBCrudPolicy:
TableName: !Ref UsersTable
- SQSSendMessagePolicy:
QueueName: !GetAtt ProcessingQueue.QueueName
- AWSLambdaBasicExecutionRole
Events:
TelegramWebhook:
Type: HttpApi
Properties:
ApiId: !Ref TelegramWebhookApi
Path: /webhook
Method: POST
################################
# Lambda2 (SQS â Lex + DynamoDB)
################################
Lambda2:
Type: AWS::Serverless::Function
Properties:
FunctionName: !Sub "${AWS::StackName}-lambda-LexBridge"
CodeUri: lambda2/
Handler: lambda_function.lambda_handler
Runtime: python3.11
Timeout: 15
DeadLetterQueue:
Type: SQS
TargetArn: !GetAtt DLQ.Arn
Environment:
Variables:
LOGS_TABLE_NAME: !Ref LogsTable
LEX_BOT_ID: !ImportValue LexBotId-Export
LEX_BOT_ALIAS_ID: !ImportValue LexBotAliasId-Export
LEX_LOCALE_ID: en_US
TELEGRAM_TOKEN: "TODO"
Policies:
- DynamoDBCrudPolicy:
TableName: !Ref LogsTable
- SQSPollerPolicy:
QueueName: !Ref ProcessingQueue
- AWSLambdaBasicExecutionRole
# Permissions for Lex are inline because there is no template in SAM.
- Version: '2012-10-17'
Statement:
- Effect: Allow
Action:
- lex:RecognizeText
Resource: "*"
Events:
SQSTrigger:
Type: SQS
Properties:
Queue: !GetAtt ProcessingQueue.Arn
BatchSize: 1
################################
# Lambda3 (Lex â Telegram)
################################
Lambda3:
Type: AWS::Serverless::Function
Properties:
FunctionName: !Sub "${AWS::StackName}-lambda-LexIntentHandler"
CodeUri: lambda3/
Handler: lambda_function.lambda_handler
Runtime: python3.11
Timeout: 15
Environment:
Variables:
OPENAI_API_KEY: #TODO
Policies:
- AWSLambdaBasicExecutionRole
################################
# Lambda Permission: Lex -> Lambda3
################################
LexInvokePermissionForLambda:
Type: AWS::Lambda::Permission
Properties:
Action: lambda:InvokeFunction
FunctionName: !Ref Lambda3
Principal: lexv2.amazonaws.com
SourceArn: !Sub
- "arn:aws:lex:${Region}:${AccountId}:bot-alias/${BotId}/${BotAliasId}"
- Region: !Ref "AWS::Region"
AccountId: !Ref "AWS::AccountId"
BotId: !ImportValue LexBotId-Export
BotAliasId: !ImportValue LexBotAliasId-Export
Deploy:
sam build
sam deploy --guided
9) Testing
Test with simple messages first: â echo test â returned You said: echo test.
Testing with AskProfessorIntent messages:
Messages like âask professorâŠâ returned a Dumbledore-style answer. đ§ââïž
5. Closing ThoughtsđȘ
This project started as a simple idea â âWhat if Dumbledore could answer my questions on Telegram?â
But connecting API Gateway, Lambda, SQS, Lex, DynamoDB, and the ChatGPT API turned into a very practical learning experience. It showed me how to design a small serverless system that stays stable, logs everything, and can grow if needed.
If you are thinking about building a chatbot, I recommend trying something fun first. When the character responds from your phone, the architecture suddenly feels alive.
Thanks for reading. I hope this write-up gives you a small push to build your own magical bot.
đ GitHub Repository
You can find the full source code here:
đ GitHub Repo:
https://github.com/sakican/serverless-lex-telegram-bot
