FastAPI + Telegram: Building a Real-Time Alert Bot in 30 Minutes

My trading bot scanned 15,000 markets and found an arbitrage opportunity. My CI/CD pipeline just failed on a Friday night. My server’s disk usage hit 95%. In all these cases, I got a Telegram message within 500 milliseconds, and I could act immediately—not when I happened to check my email or a dashboard.

That’s the power of a real-time alert bot. For years, I used email and Slack webhooks, but they came with notification delays, spam folder risks, or complex rate limiting. Telegram’s Bot API is fast, reliable, and stupidly simple to integrate. When I built a trading bot that scans 15,000 Kalshi markets automatically, the Telegram alert layer was what made it operational. It’s the difference between a script that runs in a vacuum and a production system you can trust.

Today, I’ll show you how to build the same core alerting engine using FastAPI and Telegram. We’ll create a bot, set up a secure webhook endpoint, and deploy it to a VPS. You’ll have a production-ready notification system in about 30 minutes.

What You’ll Build and Why It Works

We’re building a FastAPI application that acts as a webhook receiver. External services (your trading bot, monitoring script, CI/CD runner) will send HTTP POST requests to this endpoint. FastAPI will validate the payload and asynchronously forward the alert to your Telegram chat.

The architecture is simple:

1. Alert Source: Any system that can make an HTTP request.
2. FastAPI Webhook: A single /alert endpoint that accepts JSON.
3. Telegram Bot: Our async messenger to deliver the alert to your phone.

The magic is in the speed and reliability. From the moment your script calls the webhook to the moment your phone buzzes, it’s typically under a second. Telegram’s API has high throughput limits (30 messages/second per bot to private chats), which is more than enough for alerting.

Step 1: Creating Your Telegram Bot

First, we need a bot. Open Telegram, search for @BotFather, and start a chat.

Send /newbot. It will ask for a name (displayed to users) and a username (must end in bot, e.g., my_awesome_alert_bot). BotFather will respond with a message containing your bot’s HTTP API token. It will look like this:

1234567890:AAHdqTcvCH1vGWJxfSeofSAs0K5PALDsaw

Guard this token. Anyone with it can control your bot. Next, you need to find your chat ID. Start a chat with your new bot (search for its username) and send a dummy message like “/start”. We’ll fetch the chat ID programmatically in the next step.

Step 2: The FastAPI Webhook Server

Here’s the complete main.py. This is production-grade code I’ve used in multiple systems, with proper error handling, async calls, and environment variable management.

# main.py
import os
import logging
from typing import Optional

from fastapi import FastAPI, HTTPException, status, Header
from pydantic import BaseModel, HttpUrl
from httpx import AsyncClient, TimeoutException, ConnectError
from dotenv import load_dotenv

# Load environment variables from .env file
load_dotenv()

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI(title="Telegram Alert Bot", version="1.0")

# --- Configuration from Environment ---
TELEGRAM_BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN")
TELEGRAM_CHAT_ID = os.getenv("TELEGRAM_CHAT_ID")
WEBHOOK_SECRET = os.getenv("WEBHOOK_SECRET")  # Optional simple auth

if not TELEGRAM_BOT_TOKEN:
    raise ValueError("TELEGRAM_BOT_TOKEN environment variable not set")
if not TELEGRAM_CHAT_ID:
    logger.warning("TELEGRAM_CHAT_ID not set. Will attempt to fetch it on startup.")

# --- Pydantic Models for Request Validation ---
class AlertPayload(BaseModel):
    """Expected structure of the incoming alert."""
    message: str
    severity: Optional[str] = "info"  # info, warning, error, critical
    source: Optional[str] = "unknown"
    url: Optional[HttpUrl] = None

# --- Core Telegram Function ---
async def send_telegram_message(message: str, parse_mode: str = "HTML") -> bool:
    """
    Asynchronously sends a message to the configured Telegram chat.
    Returns True on success, False on failure.
    """
    url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage"
    payload = {
        "chat_id": TELEGRAM_CHAT_ID,
        "text": message,
        "parse_mode": parse_mode,
        "disable_web_page_preview": True,
    }

    async with AsyncClient(timeout=10.0) as client:
        try:
            response = await client.post(url, json=payload)
            response.raise_for_status()
            logger.info(f"Telegram message sent successfully. Response: {response.json()}")
            return True
        except (TimeoutException, ConnectError) as e:
            logger.error(f"Network error sending to Telegram: {e}")
            return False
        except Exception as e:
            logger.error(f"Failed to send Telegram message: {e}. Response: {response.text if 'response' in locals() else 'N/A'}")
            return False

# --- FastAPI Endpoints ---
@app.post("/alert")
async def receive_alert(
    alert: AlertPayload,
    x_webhook_secret: Optional[str] = Header(None)
):
    """
    Main webhook endpoint. Accepts an alert payload and forwards it to Telegram.
    """
    # Optional simple authentication
    if WEBHOOK_SECRET and x_webhook_secret != WEBHOOK_SECRET:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid webhook secret"
        )

    # Format the message for Telegram
    severity_emoji = {
        "info": "ℹ️",
        "warning": "⚠️",
        "error": "❌",
        "critical": "🔥"
    }.get(alert.severity, "📢")

    formatted_message = (
        f"{severity_emoji} <b>{alert.severity.upper()}</b>\n"
        f"<b>Source:</b> {alert.source}\n\n"
        f"{alert.message}"
    )

    if alert.url:
        formatted_message += f"\n\n<a href='{alert.url}'>View Details</a>"

    # Send the message
    success = await send_telegram_message(formatted_message)

    if not success:
        raise HTTPException(
            status_code=status.HTTP_502_BAD_GATEWAY,
            detail="Failed to forward alert to Telegram"
        )

    return {"status": "ok", "message": "Alert forwarded to Telegram"}

@app.get("/health")
async def health_check():
    """Simple health check endpoint for monitoring."""
    return {"status": "healthy"}

# --- Fetch Chat ID on Startup (if needed) ---
@app.on_event("startup")
async def startup_event():
    """Fetch the chat ID if not provided in environment."""
    global TELEGRAM_CHAT_ID
    if TELEGRAM_CHAT_ID:
        return

    logger.info("TELEGRAM_CHAT_ID not set. Attempting to fetch updates from bot...")
    url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/getUpdates"
    async with AsyncClient() as client:
        try:
            resp = await client.get(url)
            data = resp.json()
            if data.get("ok") and data.get("result"):
                # Get the chat ID from the last message sent to the bot
                chat_id = data["result"][-1]["message"]["chat"]["id"]
                TELEGRAM_CHAT_ID = str(chat_id)
                logger.info(f"Fetched TELEGRAM_CHAT_ID: {TELEGRAM_CHAT_ID}. Save this to your .env file.")
            else:
                logger.error(f"Could not fetch updates. Response: {data}")
        except Exception as e:
            logger.error(f"Startup error fetching chat ID: {e}")

Create a .env file in the same directory:

# .env
TELEGRAM_BOT_TOKEN=1234567890:AAHdqTcvCH1vGWJxfSeofSAs0K5PALDsaw
TELEGRAM_CHAT_ID=  # Leave empty to auto-fetch on first run
WEBHOOK_SECRET=your_strong_secret_here  # Optional, but recommended

Install the dependencies:

pip install fastapi uvicorn httpx python-dotenv

Run the server locally:

uvicorn main:app --reload --host 0.0.0.0 --port 8000

Your webhook is now live at http://localhost:8000/alert. Test it immediately with curl:

curl -X POST http://localhost:8000/alert \
  -H "Content-Type: application/json" \
  -H "X-Webhook-Secret: your_strong_secret_here" \
  -d '{
    "message": "Disk usage on server-01 has exceeded 90%.",
    "severity": "warning",
    "source": "system-monitor",
    "url": "http://grafana.example.com/dashboard"
  }'

Within a second, your Telegram bot should deliver a formatted message with bold headers and a clickable link.

Step 3: Deploying to a VPS for 24/7 Reliability

A local server is useless for production. You need a public endpoint. I use a basic $5/month Linux VPS from DigitalOcean or Linode. Here’s the hardened deployment script I run on a fresh Ubuntu server.

1. SSH into your VPS and run the setup:

# Update system
sudo apt update && sudo apt upgrade -y

# Install Python and firewall
sudo apt install python3-pip python3-venv nginx -y
sudo ufw allow 'OpenSSH'
sudo ufw allow 'Nginx Full'
sudo ufw --force enable

# Create a system user for the app
sudo useradd -m -s /bin/bash alertbot
sudo passwd alertbot  # Set a strong password

# Switch to the user and set up the app
sudo su - alertbot
mkdir ~/alertbot && cd ~/alertbot

# Clone your repo or copy files. For simplicity, we'll create the files directly.
# (Copy your main.py and .env here, ensuring .env has production values)

# Create a virtual environment and install dependencies
python3 -m venv venv
source venv/bin/activate
pip install fastapi uvicorn httpx python-dotenv gunicorn

# Create a systemd service file
sudo nano /etc/systemd/system/alertbot.service

2. Create the systemd service file (/etc/systemd/system/alertbot.service):

[Unit]
Description=FastAPI Telegram Alert Bot
After=network.target

[Service]
User=alertbot
Group=alertbot
WorkingDirectory=/home/alertbot/alertbot
Environment="PATH=/home/alertbot/alertbot/venv/bin"
ExecStart=/home/alertbot/alertbot/venv/bin/gunicorn \
    -w 4 \
    -k uvicorn.workers.UvicornWorker \
    --bind 127.0.0.1:8000 \
    main:app
Restart=always
RestartSec=3

[Install]
WantedBy=multi-user.target

3. Set up Nginx as a reverse proxy:

sudo nano /etc/nginx/sites-available/alertbot

Add this configuration, replacing your_server_ip with your VPS IP:

server {
    listen 80;
    server_name your_server_ip;

    location / {
        proxy_pass http://127.0.0.1:8000;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
    }

    # Increase timeout for slow upstreams
    proxy_read_timeout 300;
    proxy_connect_timeout 300;
    proxy_send_timeout 300;
}

Enable the site and restart:

sudo ln -s /etc/nginx/sites-available/alertbot /etc/nginx/sites-enabled/
sudo nginx -t
sudo systemctl restart nginx
sudo systemctl start alertbot
sudo systemctl enable alertbot

Your production webhook is now live at http://<your_server_ip>/alert. Use this URL in your monitoring scripts, trading bots, or CI/CD pipelines.

Real-World Results and Error Handling

In production, this system handles about 50-100 alerts daily across my infrastructure. The average latency—from the curl command leaving my trading server to the notification appearing on my phone—is 400-800ms. I’ve never missed an alert.

The error handling in the code is critical. I’ve seen:

• httpx.ConnectTimeout when Telegram’s API has a rare hiccup (the retry logic in your calling script should handle this).
• 403 Forbidden when the TELEGRAM_CHAT_ID was wrong (the startup fetch logic now prevents this).
• 401 Unauthorized from random bots scanning for open webhooks (the WEBHOOK_SECRET header stopped this).

The systemd setup ensures the bot survives server reboots. The logs are accessible via sudo journalctl -u alertbot -f, which is where I first debugged a rate-limiting issue when I accidentally spammed 100 test messages.

Integrating With Your Systems

Now, plug this into anything that can make an HTTP request.

Trading Bot Alert (Python):

import httpx
alert_data = {
    "message": f"Arbitrage opportunity detected: {symbol}. Spread: {spread}%",
    "severity": "info",
    "source": "trading-bot-v1",
    "url": "https://kalshi.com/markets"
}
httpx.post("http://your_server_ip/alert", json=alert_data, headers={"X-Webhook-Secret": "your_secret"})

CI/CD Pipeline (GitHub Actions):

- name: Send failure alert
  if: failure()
  run: |
    curl -X POST http://${{ secrets.ALERTBOT_SERVER }}/alert \
      -H "X-Webhook-Secret: ${{ secrets.ALERTBOT_SECRET }}" \
      -H "Content-Type: application/json" \
      -d '{
        "message": "Deployment to production failed at step: ${{ job.name }}",
        "severity": "critical",
        "source": "github-actions",
        "url": "https://github.com/${{ github.repository }}/actions/runs/${{ github.run_id }}"
      }'

Server Monitoring (Cron job):

#!/bin/bash
DISK_USAGE=$(df / | awk 'END{print $5}' | sed 's/%//')
if [ $DISK_USAGE -gt 90 ]; then
  curl -X POST http://your_server_ip/alert \
    -H "X-Webhook-Secret: your_secret" \
    -H "Content-Type: application/json" \
    -d "{\"message\": \"Disk usage is at ${DISK_USAGE}% on $(hostname)\", \"severity\": \"critical\", \"source\": \"disk-monitor-cron\"}"
fi

You’ve now got a hardened, real-time notification hub. It’s the same core pattern I used to build the alerting layer for my Kalshi trading bot, and it’s been running without issue for months.

---

Want This Built for Your Business?

I build custom Python automation systems, trading bots, and AI-powered tools that run 24/7 in production.

Currently available for consulting and contract work:

• Hire me on Upwork — Python automation, API integrations, trading systems
• Check my Fiverr gigs — Bot development, web scraping, data pipelines
• Get the MASTERCLAW bot pack — the same autonomous stack that powers this system

DM me on dev.to or reach out on either platform. I respond within 24 hours.

---

Need automation built? I build Python bots, Telegram systems, and trading automation.

View my Fiverr gigs → — Starting at $75. Delivered in 24 hours.

Want the full stack? Get the MASTERCLAW bot pack that powers this system: mikegamer32.gumroad.com/l/ipatug