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

At 2:47 AM, my server's CPU spiked to 98% and stayed there. I found out about it the next morning when users were already complaining. That incident cost me three hours of debugging and a genuinely embarrassing conversation with a client. Two days later, I had a Telegram bot pinging me within seconds of any anomaly. I haven't missed a critical alert since.

If you're still relying on email alerts, cron-job log checks, or — God forbid — manually SSHing into servers to verify things are running, this tutorial will change how you operate. We're building a real-time alert system using FastAPI as the webhook receiver and Telegram as the delivery channel. The same architecture I use for trading signal alerts, CI/CD pipeline notifications, and server health monitoring.

Why Telegram (Not Slack, Not Email)

Slack is expensive at scale and requires your users to have accounts. Email has unpredictable delivery delays and gets buried. PagerDuty is overkill for solo developers and small teams. Telegram is free, instant, has a dead-simple bot API, and the mobile app actually wakes you up at 3 AM. The bot API rate limits are generous (30 messages/second to different chats), and there's no per-seat pricing.

The Telegram Bot API also accepts both polling and webhooks. We're using webhooks because polling burns resources and introduces latency. With a webhook, Telegram pushes updates to your FastAPI endpoint the moment something happens.

Step 1: Create Your Bot with BotFather

Open Telegram and search for @BotFather. Send /newbot, follow the prompts, and you'll receive a token that looks like this:

5839201847:AAHxyz_example_token_here_abc123def456

Keep that token private. Anyone with it can send messages as your bot. Store it in an environment variable, not hardcoded in your source.

Next, get your chat ID. Start a conversation with your new bot (just send /start), then hit this URL in your browser, replacing YOUR_TOKEN:

https://api.telegram.org/botYOUR_TOKEN/getUpdates

The response will include your chat.id. For personal alerts, this is your personal ID. For team alerts, create a group, add the bot, send a message, and grab the group's chat ID from the same endpoint. Group IDs are negative numbers (e.g., -1001234567890).

Step 2: Project Setup

mkdir fastapi-telegram-bot && cd fastapi-telegram-bot
python -m venv venv && source venv/bin/activate
pip install fastapi uvicorn httpx python-dotenv pydantic

Create a .env file:

TELEGRAM_BOT_TOKEN=5839201847:AAHxyz_your_actual_token
TELEGRAM_CHAT_ID=123456789
WEBHOOK_SECRET=your_random_secret_string_here

The WEBHOOK_SECRET is a string you'll attach to your webhook URL so random internet traffic can't trigger your alerts. Generate it with python -c "import secrets; print(secrets.token_hex(32))".

Step 3: The FastAPI Application

Here's the full working application. I'll walk through the key decisions after the code:

import os
import logging
from contextlib import asynccontextmanager
from typing import Any

import httpx
from fastapi import FastAPI, HTTPException, Header, Request, status
from pydantic import BaseModel, Field
from dotenv import load_dotenv

load_dotenv()

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

TELEGRAM_BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN")
TELEGRAM_CHAT_ID = os.getenv("TELEGRAM_CHAT_ID")
WEBHOOK_SECRET = os.getenv("WEBHOOK_SECRET")
TELEGRAM_API_URL = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}"

if not all([TELEGRAM_BOT_TOKEN, TELEGRAM_CHAT_ID, WEBHOOK_SECRET]):
    raise RuntimeError("Missing required environment variables. Check .env file.")


class AlertPayload(BaseModel):
    title: str = Field(..., min_length=1, max_length=100)
    message: str = Field(..., min_length=1, max_length=3000)
    severity: str = Field(default="info", pattern="^(info|warning|critical)$")
    source: str = Field(default="system")
    chat_id: str | None = None  # Override default chat if needed


class TelegramSender:
    def __init__(self):
        self.client = httpx.AsyncClient(timeout=10.0)

    async def send_message(
        self,
        text: str,
        chat_id: str,
        parse_mode: str = "HTML",
        retries: int = 3
    ) -> dict[str, Any]:
        url = f"{TELEGRAM_API_URL}/sendMessage"
        payload = {
            "chat_id": chat_id,
            "text": text,
            "parse_mode": parse_mode,
            "disable_web_page_preview": True,
        }

        for attempt in range(retries):
            try:
                response = await self.client.post(url, json=payload)
                response.raise_for_status()
                result = response.json()

                if not result.get("ok"):
                    raise ValueError(f"Telegram API error: {result.get('description')}")

                logger.info(f"Message sent successfully. Message ID: {result['result']['message_id']}")
                return result

            except httpx.TimeoutException:
                logger.warning(f"Timeout on attempt {attempt + 1}/{retries}")
                if attempt == retries - 1:
                    raise
            except httpx.HTTPStatusError as e:
                # 429 = rate limited, back off and retry
                if e.response.status_code == 429:
                    retry_after = int(e.response.headers.get("Retry-After", 5))
                    logger.warning(f"Rate limited. Waiting {retry_after}s before retry.")
                    import asyncio
                    await asyncio.sleep(retry_after)
                else:
                    raise

        raise RuntimeError("Failed to send message after all retries")

    async def close(self):
        await self.client.aclose()


telegram = TelegramSender()


def format_alert_message(alert: AlertPayload) -> str:
    severity_emoji = {
        "info": "ℹ️",
        "warning": "⚠️",
        "critical": "🚨"
    }
    emoji = severity_emoji.get(alert.severity, "ℹ️")

    return (
        f"{emoji} <b>[{alert.severity.upper()}] {alert.title}</b>\n\n"
        f"{alert.message}\n\n"
        f"<i>Source: {alert.source}</i>"
    )


@asynccontextmanager
async def lifespan(app: FastAPI):
    logger.info("Starting FastAPI Telegram Alert Bot")
    yield
    logger.info("Shutting down — closing HTTP client")
    await telegram.close()


app = FastAPI(
    title="Telegram Alert Bot",
    description="Real-time alert delivery via Telegram",
    version="1.0.0",
    lifespan=lifespan
)


@app.post("/alert/{secret}", status_code=status.HTTP_200_OK)
async def receive_alert(
    secret: str,
    alert: AlertPayload,
    request: Request
):
    # Validate the secret before doing anything else
    if secret != WEBHOOK_SECRET:
        logger.warning(f"Invalid secret attempt from {request.client.host}")
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Invalid webhook secret"
        )

    chat_id = alert.chat_id or TELEGRAM_CHAT_ID
    message_text = format_alert_message(alert)

    try:
        result = await telegram.send_message(
            text=message_text,
            chat_id=chat_id
        )
        return {
            "status": "delivered",
            "message_id": result["result"]["message_id"],
            "chat_id": chat_id
        }
    except httpx.HTTPStatusError as e:
        logger.error(f"Telegram HTTP error: {e.response.status_code} - {e.response.text}")
        raise HTTPException(
            status_code=status.HTTP_502_BAD_GATEWAY,
            detail=f"Telegram API returned {e.response.status_code}"
        )
    except Exception as e:
        logger.error(f"Unexpected error sending alert: {str(e)}")
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail="Failed to deliver alert"
        )


@app.get("/health")
async def health_check():
    return {"status": "ok", "bot_configured": bool(TELEGRAM_BOT_TOKEN)}

A few decisions worth explaining: I'm using httpx.AsyncClient instead of the requests library because FastAPI is async, and blocking the event loop with synchronous HTTP calls is a real problem under load. The retry logic with 429 handling isn't theoretical — Telegram will rate-limit you if you're sending burst alerts from multiple sources simultaneously. The secret in the URL path is crude but effective; a header-based approach works too, but this is easier to configure in most CI/CD systems.

Step 4: Sending Alerts From Anywhere

Run the server locally first:

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

Test it with curl:

curl -X POST "http://localhost:8000/alert/your_secret_here" \
  -H "Content-Type: application/json" \
  -d '{
    "title": "CPU Spike Detected",
    "message": "Server cpu01 hit 97% CPU usage. Load average: 14.2, 13.8, 12.1. Top process: python3 (PID 18432).",
    "severity": "critical",
    "source": "prometheus-alertmanager"
  }'

Expected response:

{
  "status": "delivered",
  "message_id": 47,
  "chat_id": "123456789"
}

And on your phone, within about 300ms, you'll see:

🚨 [CRITICAL] CPU Spike Detected

Server cpu01 hit 97% CPU usage. Load average: 14.2, 13.8, 12.1. Top process: python3 (PID 18432).

Source: prometheus-alertmanager

For a trading alert system, I send payloads like this from a separate signal-detection service:

import httpx
import asyncio

async def send_trading_alert(signal: dict):
    alert_url = "https://your-vps.com/alert/your_secret_here"

    message = (
        f"Symbol: <b>{signal['symbol']}</b>\n"
        f"Action: <b>{signal['action']}</b>\n"
        f"Price: ${signal['price']:.4f}\n"
        f"RSI: {signal['rsi']:.1f} | MACD: {signal['macd']:.6f}\n"
        f"Confidence: {signal['confidence']}%"
    )

    payload = {
        "title": f"{signal['action']} Signal — {signal['symbol']}",
        "message": message,
        "severity": "warning" if signal['confidence'] < 75 else "critical",
        "source": "trading-engine-v2"
    }

    async with httpx.AsyncClient(timeout=5.0) as client:
        response = await client.post(alert_url, json=payload)
        response.raise_for_status()
        return response.json()

Step 5: VPS Deployment

On your Ubuntu VPS (I use a $6/month Hetzner CX11 for this), set up a systemd service so it survives reboots:

# /etc/systemd/system/telegram-alert-bot.service
[Unit]
Description=FastAPI Telegram Alert Bot
After=network.target

[Service]
Type=simple
User=ubuntu
WorkingDirectory=/home/ubuntu/fastapi-telegram-bot
EnvironmentFile=/home/ubuntu/fastapi-telegram-bot/.env
ExecStart=/home/ubuntu/fastapi-telegram-bot/venv/bin/uvicorn main:app --host 0.0.0.0 --port 8000 --workers 2
Restart=always
RestartSec=5

[Install]
WantedBy=multi-user.target

sudo systemctl daemon-reload
sudo systemctl enable telegram-alert-bot
sudo systemctl start telegram-alert-bot
sudo systemctl status telegram-alert-bot

Put Nginx in front of it for SSL termination. Without HTTPS, your webhook secret is traveling in plaintext. Get a free cert with Certbot and point your Nginx location block to proxy_pass http://127.0.0.1:8000.

For server monitoring, I run a simple cron job that hits the alert endpoint if disk or memory crosses a threshold:

# crontab -e
*/5 * * * * /home/ubuntu/scripts/check_health.sh

For CI/CD, add a step to your GitHub Actions workflow:

- name: Send deployment alert
  run: |
    curl -s -X POST "${{ secrets.ALERT_WEBHOOK_URL }}/alert/${{ secrets.WEBHOOK_SECRET }}" \
      -H "Content-Type: application/json" \
      -d "{\"title\": \"Deployment Complete\", \"message\": \"${{ github.repository }} deployed commit ${{ github.sha }} to production.\", \"severity\": \"info\", \"source\": \"github-actions\"}"

Real-World Performance Numbers

In production, this setup handles about 2,000 alerts/day across three different projects without any issues. End-to-end latency from the triggering event to Telegram notification appearing on my phone averages 380ms on a good network day, occasionally spiking to 1.2 seconds if Telegram's servers are having a moment. The FastAPI endpoint itself responds in under 5ms — all the latency is in the outbound HTTP call to Telegram.

The one error I hit repeatedly when first building this was {"ok":false,"error_code":400,"description":"Bad Request: chat not found"}. This almost always means your bot hasn't been started by the user, or for groups, the bot wasn't added as a member before you tried to message it. Send /start to your bot first, or verify the group ID is correct.

Memory usage for the service sits at about 45MB RSS with two Uvicorn workers — completely negligible. This is the kind of infrastructure that runs forever without you thinking about it, which is exactly what you want from an alerting system.

---

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

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