Оффлайн
- Регистрация
- 9 Май 2015
- Сообщения
- 1,480
- Баллы
- 155
In this guide, you’ll build a monitoring workflow using FastAPI (Python) for checks, n8n as the orchestration/automation, and Telegram for immediate alerts.
Here's what you'll build
Prerequisites & Tools
N8N is a no-code workflow automation tool that allows you to connect to different apps and services easily. With n8n, you can connect different apps using nodes that represent a specific action, such as connecting to an API, sending an email, or scraping a website. Typically, an n8n workflow includes
Setting up the project
1. Project Layout
Files you'd need to create.
monitoring/
├── python_service/
│ ├── app.py
│ ├── monitor.py
│ ├── requirements.txt
│ └── Dockerfile
├── n8n/
│ └── workflows/telegram_monitor.json
├── docker-compose.yml
└── .env
2. Python Microservice (FastAPI)
Create python_service/app.py:
from fastapi import FastAPI
from pydantic import BaseModel
import time
from typing import List
from monitor import perform_checks, CheckInput, CheckResult, load_state, save_state
app = FastAPI(title="Monitoring Service")
class CheckInputList(BaseModel):
endpoints: List[CheckInput]
class CheckResultList(BaseModel):
results: List[CheckResult]
@app.on_event("startup")
def startup_event():
# load persistent state into monitor
load_state()
@app.post("/check", response_model=CheckResultList)
def check_all(ci: CheckInputList):
results = perform_checks(ci.endpoints)
save_state()
return {"results": results}
@app.get("/health")
def health():
return {"status": "ok", "time": time.time()}
Create python_service/monitor.py:
import os
import json
import time
from typing import List, Optional
import httpx
from pydantic import BaseModel
STATE_PATH = os.getenv("STATE_PATH", "/data/state.json")
class CheckInput(BaseModel):
name: Optional[str]
url: str
expect_text: Optional[str] = None
class CheckResult(BaseModel):
name: Optional[str]
url: str
ok: bool
status_code: Optional[int] = None
latency_ms: Optional[int] = None
error: Optional[str] = None
transitioned: Optional[str] = None # "up_to_down", "down_to_up", or None
time: float = time.time()
_state = {} # url -> {"ok": bool}
def load_state():
global _state
try:
with open(STATE_PATH, "r") as f:
_state = json.load(f)
except Exception:
_state = {}
def save_state():
global _state
try:
os.makedirs(os.path.dirname(STATE_PATH), exist_ok=True)
with open(STATE_PATH, "w") as f:
json.dump(_state, f)
except Exception as e:
print("Failed to save state:", e)
def perform_check(ci: CheckInput) -> CheckResult:
start = time.monotonic()
resp = None
try:
resp = httpx.get(ci.url, timeout=10)
latency = int((time.monotonic() - start) * 1000)
ok = (resp.status_code == 200) and (ci.expect_text is None or ci.expect_text in resp.text)
err = None
except Exception as e:
ok = False
latency = None
err = str(e)
prev = _state.get(ci.url, {"ok": True})
transitioned = None
if ok and not prev.get("ok", True):
transitioned = "down_to_up"
elif not ok and prev.get("ok", True):
transitioned = "up_to_down"
_state[ci.url] = {"ok": ok}
return CheckResult(
name=ci.name,
url=ci.url,
ok=ok,
status_code=(resp.status_code if resp else None),
latency_ms=latency,
error=err,
transitioned=transitioned,
time=time.time(),
)
def perform_checks(endpoints: List[CheckInput]) -> List[CheckResult]:
results = []
for ep in endpoints:
results.append(perform_check(ep))
return results
requirements.txt:
fastapi
uvicorn[standard]
httpx
pydantic
Dockerfile:
FROM python:3.10-slim
WORKDIR /app
COPY requirements.txt ./
RUN pip install --no-cache-dir -r requirements.txt
COPY app.py monitor.py ./
ENV STATE_PATH=/data/state.json
RUN mkdir -p /data
EXPOSE 8000
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"]
3. The docker-compose.yml
Next, we'll be creating the docker-compose.yml file:
version: "3.8"
services:
n8n:
image: n8nio/n8n:latest
environment:
- DB_SQLITE_VACUUM_ON_STARTUP=true
- N8N_BASIC_AUTH_ACTIVE=true
- N8N_BASIC_AUTH_USER=${N8N_USER}
- N8N_BASIC_AUTH_PASSWORD=${N8N_PASSWORD}
- WEBHOOK_URL=${WEBHOOK_URL}
ports:
- "5679:5678"
depends_on:
- python
networks:
- monitor_net
python:
build:
context: ./python_service
volumes:
- python_state:/data
networks:
- monitor_net
networks:
monitor_net:
driver: bridge
volumes:
python_state:
Also, let's create config.json
{
"auths": {
"": {},
"": {},
"": {}
},
"credsStore": "wincred",
"currentContext": "desktop-linux"
}
.env (example):
N8N_USER=monitorapp
N8N_PASSWORD='Eas8Y!P@@sw@rd'
WEBHOOK_URL=
N8N_PORT=5679 # optional host port for n8n
Finally, create telegram_monitor.json:
{
"nodes": [
{
"parameters": {
"interval": 1,
"unit": "minutes"
},
"name": "Schedule Trigger",
"type": "n8n-nodes-base.scheduleTrigger",
"typeVersion": 1,
"position": [250, 300]
},
{
"parameters": {
"url": "",
"method": "POST",
"jsonParameters": true,
"bodyParametersJson": "={\"endpoints\": [ {\"name\":\"prod\",\"url\":\"https://www.invalid.domain\",\"expect_text\":\"OK\"} ] }"
},
"name": "Python Check",
"type": "n8n-nodes-base.httpRequest",
"typeVersion": 2,
"position": [450, 300]
},
{
"parameters": {
"value": "={{ $node[\"Python Check\"].json[\"results\"] }}",
"options": {
"type": "splitInBatches",
"batchSize": 1
}
},
"name": "Split Results",
"type": "n8n-nodes-base.splitInBatches",
"typeVersion": 1,
"position": [650, 300]
},
{
"parameters": {
"rules": [
{
"value1": "={{ $json[\"transitioned\"] }}",
"operation": "isNotEmpty"
}
]
},
"name": "Filter Transitions",
"type": "n8n-nodes-base.if",
"typeVersion": 1,
"position": [850, 300]
},
{
"parameters": {
"chatId": "98765432",
"text": "={{ $json[\"name\"] || $json[\"url\"] }} changed status: {{ $json[\"transitioned\"] }}. StatusCode={{ $json[\"status_code\"] }}, Latency={{ $json[\"latency_ms\"] }}ms, Error={{ $json[\"error\"] }}",
"parseMode": "HTML"
},
"name": "Telegram Alert",
"type": "n8n-nodes-base.telegram",
"typeVersion": 1,
"position": [1050, 300]
}
],
"connections": {
"Schedule Trigger": {
"main": [
[
{
"node": "Python Check",
"type": "main",
"index": 0
}
]
]
},
"Python Check": {
"main": [
[
{
"node": "Split Results",
"type": "main",
"index": 0
}
]
]
},
"Split Results": {
"main": [
[
{
"node": "Filter Transitions",
"type": "main",
"index": 0
}
]
]
},
"Filter Transitions": {
"main": [
[
{
"node": "Telegram Alert",
"type": "main",
"index": 0
}
]
]
}
}
}
4. Building the n8n workflow
We can choose to build the workflow manually or import the JSON file. Since we'll be running n8n locally rather than cloud, n8n recommends using Docker for most self-hosting needs. You can also use n8n in Docker with Docker Compose.
Starting n8n
From your terminal, run the following commands, replacing the placeholders with your timezone:
docker volume create n8n_data
docker run -it --rm \
--name n8n \
-p 5678:5678 \
-e GENERIC_TIMEZONE="<YOUR_TIMEZONE>" \
-e TZ="<YOUR_TIMEZONE>" \
-e N8N_ENFORCE_SETTINGS_FILE_PERMISSIONS=true \
-e N8N_RUNNERS_ENABLED=true \
-v n8n_data:/home/node/.n8n \
docker.n8n.io/n8nio/n8n
Once it's running, you can access n8n at .
6. Configure the Telegram Node
Status Update
Service: {{$json["results"] ? ($json["results"][0]["name"] || $json["results"][0]["url"]) : ($json["name"] || $json["url"] || "Unknown")}}
Status: {{$json["results"] ? ($json["results"][0]["ok"] ? "UP
" : "DOWN 
") : ($json["ok"] ? "UP " : "DOWN ")}}
Status Code: {{$json["results"] ? ($json["results"][0]["status_code"] || "N/A") : ($json["status_code"] || "N/A")}}
Latency: {{$json["results"] ? ($json["results"][0]["latency_ms"] + " ms") : ($json["latency_ms"] + " ms") || "N/A"}}
Error: {{$json["results"] ? ($json["results"][0]["error"] || "None") : ($json["error"] || "None")}}
7. Configure Python Check
8. Test the workflow
You’ve now successfully built a fully automated monitoring pipeline using Python and n8n with real-time alerts and total control over your automation.
You can always swap your Telegram node with Slack/Email node, monitor multiple environments and even track your last alert timestamp.
Excited to see what you build!
Here's what you'll build
- A Python microservice (FastAPI) that performs endpoint checks /check and exposes /health.
- n8n workflow that calls the Python service on a schedule, filters results, and sends well-formatted Telegram alerts.
- Docker Compose deployment; everything runs in containers on the same network.
Prerequisites & Tools
- Python (version python:3.10-slim)
- Docker & Docker Compose
- n8n (self-hosted)
- Telegram Bot / Chat ID
- Basic knowledge of HTTP, JSON
N8N is a no-code workflow automation tool that allows you to connect to different apps and services easily. With n8n, you can connect different apps using nodes that represent a specific action, such as connecting to an API, sending an email, or scraping a website. Typically, an n8n workflow includes
Trigger node: Starts your workflow (e.g, Gmail trigger, webhook, etc).
Action nodes: Perform actions in your workflow and define specific applications (e.g, organizing data, sending messages, calling APIs, etc).
Setting up the project
1. Project Layout
Files you'd need to create.
monitoring/
├── python_service/
│ ├── app.py
│ ├── monitor.py
│ ├── requirements.txt
│ └── Dockerfile
├── n8n/
│ └── workflows/telegram_monitor.json
├── docker-compose.yml
└── .env
2. Python Microservice (FastAPI)
Create python_service/app.py:
from fastapi import FastAPI
from pydantic import BaseModel
import time
from typing import List
from monitor import perform_checks, CheckInput, CheckResult, load_state, save_state
app = FastAPI(title="Monitoring Service")
class CheckInputList(BaseModel):
endpoints: List[CheckInput]
class CheckResultList(BaseModel):
results: List[CheckResult]
@app.on_event("startup")
def startup_event():
# load persistent state into monitor
load_state()
@app.post("/check", response_model=CheckResultList)
def check_all(ci: CheckInputList):
results = perform_checks(ci.endpoints)
save_state()
return {"results": results}
@app.get("/health")
def health():
return {"status": "ok", "time": time.time()}
Create python_service/monitor.py:
import os
import json
import time
from typing import List, Optional
import httpx
from pydantic import BaseModel
STATE_PATH = os.getenv("STATE_PATH", "/data/state.json")
class CheckInput(BaseModel):
name: Optional[str]
url: str
expect_text: Optional[str] = None
class CheckResult(BaseModel):
name: Optional[str]
url: str
ok: bool
status_code: Optional[int] = None
latency_ms: Optional[int] = None
error: Optional[str] = None
transitioned: Optional[str] = None # "up_to_down", "down_to_up", or None
time: float = time.time()
_state = {} # url -> {"ok": bool}
def load_state():
global _state
try:
with open(STATE_PATH, "r") as f:
_state = json.load(f)
except Exception:
_state = {}
def save_state():
global _state
try:
os.makedirs(os.path.dirname(STATE_PATH), exist_ok=True)
with open(STATE_PATH, "w") as f:
json.dump(_state, f)
except Exception as e:
print("Failed to save state:", e)
def perform_check(ci: CheckInput) -> CheckResult:
start = time.monotonic()
resp = None
try:
resp = httpx.get(ci.url, timeout=10)
latency = int((time.monotonic() - start) * 1000)
ok = (resp.status_code == 200) and (ci.expect_text is None or ci.expect_text in resp.text)
err = None
except Exception as e:
ok = False
latency = None
err = str(e)
prev = _state.get(ci.url, {"ok": True})
transitioned = None
if ok and not prev.get("ok", True):
transitioned = "down_to_up"
elif not ok and prev.get("ok", True):
transitioned = "up_to_down"
_state[ci.url] = {"ok": ok}
return CheckResult(
name=ci.name,
url=ci.url,
ok=ok,
status_code=(resp.status_code if resp else None),
latency_ms=latency,
error=err,
transitioned=transitioned,
time=time.time(),
)
def perform_checks(endpoints: List[CheckInput]) -> List[CheckResult]:
results = []
for ep in endpoints:
results.append(perform_check(ep))
return results
requirements.txt:
fastapi
uvicorn[standard]
httpx
pydantic
Dockerfile:
FROM python:3.10-slim
WORKDIR /app
COPY requirements.txt ./
RUN pip install --no-cache-dir -r requirements.txt
COPY app.py monitor.py ./
ENV STATE_PATH=/data/state.json
RUN mkdir -p /data
EXPOSE 8000
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"]
3. The docker-compose.yml
Next, we'll be creating the docker-compose.yml file:
version: "3.8"
services:
n8n:
image: n8nio/n8n:latest
environment:
- DB_SQLITE_VACUUM_ON_STARTUP=true
- N8N_BASIC_AUTH_ACTIVE=true
- N8N_BASIC_AUTH_USER=${N8N_USER}
- N8N_BASIC_AUTH_PASSWORD=${N8N_PASSWORD}
- WEBHOOK_URL=${WEBHOOK_URL}
ports:
- "5679:5678"
depends_on:
- python
networks:
- monitor_net
python:
build:
context: ./python_service
volumes:
- python_state:/data
networks:
- monitor_net
networks:
monitor_net:
driver: bridge
volumes:
python_state:
Also, let's create config.json
{
"auths": {
"": {},
"": {},
"": {}
},
"credsStore": "wincred",
"currentContext": "desktop-linux"
}
.env (example):
N8N_USER=monitorapp
N8N_PASSWORD='Eas8Y!P@@sw@rd'
WEBHOOK_URL=
N8N_PORT=5679 # optional host port for n8n
Finally, create telegram_monitor.json:
{
"nodes": [
{
"parameters": {
"interval": 1,
"unit": "minutes"
},
"name": "Schedule Trigger",
"type": "n8n-nodes-base.scheduleTrigger",
"typeVersion": 1,
"position": [250, 300]
},
{
"parameters": {
"url": "",
"method": "POST",
"jsonParameters": true,
"bodyParametersJson": "={\"endpoints\": [ {\"name\":\"prod\",\"url\":\"https://www.invalid.domain\",\"expect_text\":\"OK\"} ] }"
},
"name": "Python Check",
"type": "n8n-nodes-base.httpRequest",
"typeVersion": 2,
"position": [450, 300]
},
{
"parameters": {
"value": "={{ $node[\"Python Check\"].json[\"results\"] }}",
"options": {
"type": "splitInBatches",
"batchSize": 1
}
},
"name": "Split Results",
"type": "n8n-nodes-base.splitInBatches",
"typeVersion": 1,
"position": [650, 300]
},
{
"parameters": {
"rules": [
{
"value1": "={{ $json[\"transitioned\"] }}",
"operation": "isNotEmpty"
}
]
},
"name": "Filter Transitions",
"type": "n8n-nodes-base.if",
"typeVersion": 1,
"position": [850, 300]
},
{
"parameters": {
"chatId": "98765432",
"text": "={{ $json[\"name\"] || $json[\"url\"] }} changed status: {{ $json[\"transitioned\"] }}. StatusCode={{ $json[\"status_code\"] }}, Latency={{ $json[\"latency_ms\"] }}ms, Error={{ $json[\"error\"] }}",
"parseMode": "HTML"
},
"name": "Telegram Alert",
"type": "n8n-nodes-base.telegram",
"typeVersion": 1,
"position": [1050, 300]
}
],
"connections": {
"Schedule Trigger": {
"main": [
[
{
"node": "Python Check",
"type": "main",
"index": 0
}
]
]
},
"Python Check": {
"main": [
[
{
"node": "Split Results",
"type": "main",
"index": 0
}
]
]
},
"Split Results": {
"main": [
[
{
"node": "Filter Transitions",
"type": "main",
"index": 0
}
]
]
},
"Filter Transitions": {
"main": [
[
{
"node": "Telegram Alert",
"type": "main",
"index": 0
}
]
]
}
}
}
4. Building the n8n workflow
We can choose to build the workflow manually or import the JSON file. Since we'll be running n8n locally rather than cloud, n8n recommends using Docker for most self-hosting needs. You can also use n8n in Docker with Docker Compose.
Starting n8n
You can follow along with n8n's video guide here:
From your terminal, run the following commands, replacing the placeholders with your timezone:
docker volume create n8n_data
docker run -it --rm \
--name n8n \
-p 5678:5678 \
-e GENERIC_TIMEZONE="<YOUR_TIMEZONE>" \
-e TZ="<YOUR_TIMEZONE>" \
-e N8N_ENFORCE_SETTINGS_FILE_PERMISSIONS=true \
-e N8N_RUNNERS_ENABLED=true \
-v n8n_data:/home/node/.n8n \
docker.n8n.io/n8nio/n8n
Once it's running, you can access n8n at .
- Building the n8n & Telegram workflow
Open Telegram, search for (Telegram’s tool for creating and managing bots).
Use the /newbot command to create a new bot (pick a name and username ending with bot).
BotFather will give you an API token (save this to ass in your n8n credentials).
In Telegram, search for @get_id_bot or @myidbot and start a chat with it.
Send /getid to it.
It will respond to you with your chat’s ID.
Make sure n8n is running in Docker
Open htttp://localhost:5678
Log in with the N8N_USER/N8N_PASSWORD you set in .env
In n8n, go to Credentials and create new.
Choose Telegram API
Paste the BotFather token
Name the credential anything you like and save.
In n8n, click Create Workflow, and Import from File
Select monitoring/n8n/workflows/telegram_monitor.json
It will show your workflow with nodes: Schedule Trigger, Python Check, Split Results, Filter Transitions and Telegram Alert
6. Configure the Telegram Node
- Open the Telegram Alert node
- In Credentials, select the Telegram bot credential you created
- In Chat ID, replace Chat ID with the number you got from @get_id_bot or @myidbot
- Change Text to:
Status UpdateService: {{$json["results"] ? ($json["results"][0]["name"] || $json["results"][0]["url"]) : ($json["name"] || $json["url"] || "Unknown")}}
Status: {{$json["results"] ? ($json["results"][0]["ok"] ? "UP
" : "DOWN ") : ($json["ok"] ? "UP " : "DOWN ")}}Status Code: {{$json["results"] ? ($json["results"][0]["status_code"] || "N/A") : ($json["status_code"] || "N/A")}}
Latency: {{$json["results"] ? ($json["results"][0]["latency_ms"] + " ms") : ($json["latency_ms"] + " ms") || "N/A"}}
Error: {{$json["results"] ? ($json["results"][0]["error"] || "None") : ($json["error"] || "None")}}
7. Configure Python Check
8. Test the workflow
Execute the workflow
n8n will call your Python service and get results back
If a monitored site transitions, it'll send you a Telegram message according to the parameters you set.
- Finally, you should get a response like the above (make sure to edit parameters, including URL to monitor)
You’ve now successfully built a fully automated monitoring pipeline using Python and n8n with real-time alerts and total control over your automation.
You can always swap your Telegram node with Slack/Email node, monitor multiple environments and even track your last alert timestamp.
Excited to see what you build!