Chapter 12 · Deployment and Capstone: A Bot That Stays Up
A deployed bot is the same code in a place you are not sitting at. What changes is everything around it: settings that must not live in the source, a database that holds other people's conversations, a process that must start by itself, restart after a crash, and stop without losing a reply, and a path from the internet that is encrypted. This chapter assembles the twelve chapters into
one service and prepares it for a Linux server. It also closes the gap chapter 9 left open: Telegram's developer terms ask for user data to be encrypted at rest, and until now the SQLite file was plain.
What was run, and what was not (read this first)
Run here, on Windows with Python 3.14.0: the whole service as a real process (py -m bot.app) in production mode with the echo provider; a live smoke test against it; the encryption, rotation and backup code; the configuration checks; the shutdown sequence (in-process). The project now has 290 passing tests.
Run here, narrowly:systemd-analyze verify (systemd 252, Debian 12 under WSL) on the three unit files. It checks syntax and directive names, not that the service works.
Not run here: the Docker build (Docker Desktop's engine was not running and I did not start it), nginx, TLS certificates, a real systemd start, a real SIGTERM, a real provider or any chat platform. Those files are written from documentation and general knowledge and are labelled again where they appear. Treat them as a starting point to test on your own server, not as verified.
No server of yours was touched and nothing was installed in your Linux distributions.
One Service, Configured From the Environment
Until now each channel had its own small script. A deployment wants one process that starts everything that is configured: the web chat and the Slack route on the web server, Telegram polling (or its webhook), the Discord connection, plus two background jobs (the retention purge from chapter 6 and the alert check from chapter 11).
All of it is driven by environment variables, validated at start-up. The rules, in order of how much grief they save:
Secrets never go in the code, the Docker image or git. They arrive in the environment (here from a file only root can read, /etc/chatbot/chatbot.env).
Fail fast. A bad configuration should stop the start-up with a clear message, not surface three days later as a cookie that cannot be signed. In production the check demands every secret and refuses the fake provider.
Report every problem at once, not one per attempt.
Never print a secret. The summary shows set or missing.
"""All configuration comes from environment variables, is checked at start-up, and is never printed with its secrets."""
import os
import pathlib
from dataclasses import dataclass, field
PROVIDERS = {
"fake": ("fake", "", "", ""),
"echo": ("echo", "", "", ""),
"claude": ("claude", "", "", "ANTHROPIC_API_KEY"),
"grok": ("openai-compatible", "", "https://api.x.ai/v1", "XAI_API_KEY"),
"openai": ("openai-compatible", "", "https://api.openai.com/v1", "OPENAI_API_KEY"),
"ollama": ("openai-compatible", "", "http://localhost:11434/v1", ""),
}
@dataclass(frozen=True)
class Settings:
env: str = "development"
data_dir: pathlib.Path = pathlib.Path("data")
host: str = "127.0.0.1"
port: int = 8000
provider: str = "fake"
model: str = ""
base_url: str = ""
web_secret: str = ""
allowed_origins: tuple[str, ...] = ()
secure_cookie: bool = False
content_keys: tuple[str, ...] = ()
log_key: str = ""
log_level: str = "INFO"
retention_days: int = 30
persona_file: str = "persona.toml"
telegram_token: str = ""
telegram_webhook_secret: str = ""
discord_token: str = ""
slack_token: str = ""
slack_signing_secret: str = ""
problems: tuple[str, ...] = field(default=(), compare=False)
@property
def production(self) -> bool:
return self.env == "production"
@property
def db_path(self) -> str:
return str(self.data_dir / "chatbot.db")
def summary(self) -> dict[str, str]:
"""What is configured, with every secret shown only as 'set' or 'missing'."""
def flag(value) -> str:
return "set" if value else "missing"
return {
"environment": self.env, "listen": f"{self.host}:{self.port}", "provider": self.provider, "model": self.model or "(provider default)",
"data directory": str(self.data_dir), "retention days": str(self.retention_days),
"content encryption": f"{len(self.content_keys)} key(s)" if self.content_keys else "OFF (plain text on disk)",
"web session secret": flag(self.web_secret), "allowed origins": ", ".join(self.allowed_origins) or "(none)",
"log pseudonym key": flag(self.log_key), "secure cookie": str(self.secure_cookie),
"telegram": ("webhook" if self.telegram_webhook_secret else "polling") if self.telegram_token else "off",
"discord": "on" if self.discord_token else "off",
"slack": "on" if (self.slack_token and self.slack_signing_secret) else "off",
}
def load_settings(env: dict[str, str] | None = None) -> Settings:
"""Read the environment and collect every problem in one pass, so a bad configuration is fixed in one go."""
e = dict(os.environ if env is None else env)
problems: list[str] = []
def get(name: str, default: str = "") -> str:
return e.get(name, default).strip()
def number(name: str, default: int) -> int:
try:
return int(get(name, str(default)))
except ValueError:
problems.append(f"{name} must be a whole number")
return default
environment = get("CHATBOT_ENV", "development")
if environment not in ("development", "production"):
problems.append("CHATBOT_ENV must be 'development' or 'production'")
provider = get("CHATBOT_PROVIDER", "fake")
if provider not in PROVIDERS:
problems.append(f"CHATBOT_PROVIDER must be one of {', '.join(PROVIDERS)}")
else:
key_env = PROVIDERS[provider][3]
if key_env and not get(key_env):
problems.append(f"{key_env} is required for provider '{provider}'")
if provider not in ("fake", "echo") and not get("CHATBOT_MODEL"):
problems.append("CHATBOT_MODEL is required: read the current model name from your provider's model list")
keys = tuple(k.strip() for k in get("CHATBOT_CONTENT_KEYS").split(",") if k.strip())
web_secret, log_key = get("CHATBOT_WEB_SECRET"), get("CHATBOT_LOG_KEY")
origins = tuple(o.strip() for o in get("CHATBOT_ALLOWED_ORIGINS").split(",") if o.strip())
if environment == "production":
if not keys:
problems.append("CHATBOT_CONTENT_KEYS is required in production (encrypts messages on disk; generate one with: py -m bot.crypto_keygen)")
if len(web_secret) < 32:
problems.append("CHATBOT_WEB_SECRET must be at least 32 characters in production")
if len(log_key) < 16:
problems.append("CHATBOT_LOG_KEY must be at least 16 characters in production")
if not origins:
problems.append("CHATBOT_ALLOWED_ORIGINS is required in production (for example https://chat.example.org)")
if provider == "fake":
problems.append("CHATBOT_PROVIDER=fake is not allowed in production")
for key in keys:
try:
from cryptography.fernet import Fernet
Fernet(key.encode())
except Exception: # noqa: BLE001 - any failure to build a Fernet from the text means it is not a valid key
problems.append("CHATBOT_CONTENT_KEYS contains a value that is not a valid Fernet key")
break
if bool(get("SLACK_BOT_TOKEN")) != bool(get("SLACK_SIGNING_SECRET")):
problems.append("SLACK_BOT_TOKEN and SLACK_SIGNING_SECRET must be set together")
return Settings(
env=environment, data_dir=pathlib.Path(get("CHATBOT_DATA_DIR", "data")), host=get("CHATBOT_HOST", "127.0.0.1"),
port=number("CHATBOT_PORT", 8000), provider=provider, model=get("CHATBOT_MODEL"), base_url=get("CHATBOT_BASE_URL"),
web_secret=web_secret or ("" if environment == "production" else "development-only-secret-do-not-use-in-production"),
allowed_origins=origins or (() if environment == "production" else ("http://127.0.0.1:8000",)),
secure_cookie=get("CHATBOT_SECURE_COOKIE", "1" if environment == "production" else "0") == "1", content_keys=keys,
log_key=log_key or ("" if environment == "production" else "development-log-key"), log_level=get("CHATBOT_LOG_LEVEL", "INFO").upper(),
retention_days=number("CHATBOT_RETENTION_DAYS", 30), persona_file=get("CHATBOT_PERSONA_FILE", "persona.toml"),
telegram_token=get("TELEGRAM_BOT_TOKEN"), telegram_webhook_secret=get("TELEGRAM_WEBHOOK_SECRET"),
discord_token=get("DISCORD_BOT_TOKEN"), slack_token=get("SLACK_BOT_TOKEN"), slack_signing_secret=get("SLACK_SIGNING_SECRET"),
problems=tuple(problems),
)
The same program, in three situations. This is real output (the keys were generated for the demonstration, the Claude key and Telegram token are made up):
$ py -m bot.app --check (development, nothing set)
environment development
listen 127.0.0.1:8000
provider fake
model (provider default)
data directory data
retention days 30
content encryption OFF (plain text on disk)
web session secret set
allowed origins http://127.0.0.1:8000
log pseudonym key set
secure cookie False
telegram off
discord off
slack off
exit code: 0
$ CHATBOT_ENV=production py -m bot.app --check (nothing else set)
environment production
listen 127.0.0.1:8000
provider fake
model (provider default)
data directory data
retention days 30
content encryption OFF (plain text on disk)
web session secret missing
allowed origins (none)
log pseudonym key missing
secure cookie True
telegram off
discord off
slack off
PROBLEM: CHATBOT_CONTENT_KEYS is required in production (encrypts messages on disk; generate one with: py -m bot.crypto_keygen)
PROBLEM: CHATBOT_WEB_SECRET must be at least 32 characters in production
PROBLEM: CHATBOT_LOG_KEY must be at least 16 characters in production
PROBLEM: CHATBOT_ALLOWED_ORIGINS is required in production (for example https://chat.example.org)
PROBLEM: CHATBOT_PROVIDER=fake is not allowed in production
exit code: 1
$ a complete production configuration (keys generated just now, values not shown)
environment production
listen 127.0.0.1:8000
provider claude
model example-model
data directory data
retention days 30
content encryption 1 key(s)
web session secret set
allowed origins https://chat.example.org
log pseudonym key set
secure cookie True
telegram polling
discord off
slack off
exit code: 0
Note the middle run. In production, a missing web secret is a problem; in development it gets a default, which is the dangerous kind of default, so the production summary must not claim it is "set". My first version did exactly that: --check in production reported the development defaults as set, because the defaults were filled in before the summary was built. Reading the output instead of just the exit code found it.
Encrypting Messages on Disk
Chapter 9 recorded that Telegram's Developer Terms ask for user data to be encrypted at rest and that the SQLite file was plain. There are two ways to close that, and they protect against different things:
Disk or volume encryption (LUKS, a cloud volume)
Encrypting the content in the application
Protects against
A stolen disk or an old discarded drive
A copied database file, a leaked backup, someone browsing the data directory
Does not protect against
Anyone who can read files while the machine is running, including backups copied off it
Someone who gets the running process, or the key
Effort
Server setup; no code
Code (below), plus key management
I used the second, and you can still add the first; they do not conflict. The text of messages, summaries and remembered notes is encrypted with Fernet from the cryptography package, which is authenticated encryption (it detects tampering) with a simple interface. What I know about Fernet comes from the library's documented design and general knowledge; I did not re-read its specification for this chapter.
MultiFernet holds several keys: it encrypts with the first and tries all of them to decrypt, which is what makes key rotation possible.
"""Encrypting what people said before it is written to disk, with a key that is kept somewhere else."""
from cryptography.fernet import Fernet, InvalidToken, MultiFernet
TOKEN_PREFIX = "gAAAA" # every Fernet token starts like this (a version byte, base64-encoded), which lets us tell encrypted from plain
class WrongKey(Exception):
"""The stored text cannot be decrypted with the keys we hold (wrong or missing key, or the file was altered)."""
def generate_key() -> str:
return Fernet.generate_key().decode()
class ContentCipher:
"""Fernet (AES-128-CBC with an HMAC-SHA256 check, per the cryptography library's documentation) with key rotation.
Pass several keys, newest first: new text is encrypted with the first, and any listed key can still decrypt. To retire an
old key, add the new one at the front, run Store.rotate(), then remove the old one."""
def __init__(self, keys: list[str]) -> None:
if not keys:
raise ValueError("at least one key is required")
self._fernet = MultiFernet([Fernet(k.encode() if isinstance(k, str) else k) for k in keys])
def encrypt(self, text: str) -> str:
return self._fernet.encrypt(text.encode()).decode()
def decrypt(self, token: str) -> str:
try:
return self._fernet.decrypt(token.encode()).decode()
except InvalidToken:
raise WrongKey("stored text could not be decrypted with the configured key(s)") from None
def rotate(self, token: str) -> str:
"""Re-encrypt a token under the newest key."""
try:
return self._fernet.rotate(token.encode()).decode()
except InvalidToken:
raise WrongKey("stored text could not be decrypted with the configured key(s)") from None
@staticmethod
def is_encrypted(value: str) -> bool:
return value.startswith(TOKEN_PREFIX)
The store seals text on the way in and opens it on the way out; a value that is not encrypted passes through unchanged, so a database can be switched over gradually. The real output of switching an existing database over, backing it up, rotating the key and retiring the old one:
1. A database that was running WITHOUT encryption
'Leeds' readable in the file: True
2. Switch encryption on and encrypt what is already there
values encrypted: 2 (a second run: 0 )
'Leeds' readable in the file: False (VACUUM removed the old copy)
the bot still reads it: My card went missing in Leeds
3. Back it up while open
backup file: chatbot-20261010-162934.db 'Leeds' readable in the backup: False
4. Rotate: new key first, old key kept so old rows still open
rows re-encrypted under the new key: 2
5. Retire the old key
readable with only the new key: My card went missing in Leeds
the OLD backup still needs the OLD key:
WrongKey raised: stored text could not be decrypted with the configured key(s)
6. Start with the wrong key
WrongKey raised: stored text could not be decrypted with the configured key(s)
And the actual bytes of the database written by the running production process during the smoke test below. The rows are tokens starting gAAAA; the question and the reply are nowhere in the file:
plain text 'Say hello in five words' in the file: False
plain text 'You said' in the file: False
row 1 user gAAAAABqymcJQzkIN3O62zZ3wfh-BKrYX6CJb2j6X2Yl... (120 chars)
row 2 assistant gAAAAABqymcJEF-hZHadGLjDlH0A656-91L6R28DB7PV... (140 chars)
row 3 user gAAAAABqymcPR5SG7K2CVvIprG1n1S_7Xq0Vcu9PymsV... (120 chars)
row 4 assistant gAAAAABqymcPOTgCWd441vrHwdMQqtCJVMkRItcBokEG... (140 chars)
readable columns beside them: ['id', 'session_id', 'role', 'content', 'created_at']
What this does and does not give you
The key is the whole protection. Keep it out of the data directory and out of backups. A backup stored beside the key is not protected at all. Lose the key and the messages are gone for good: that is deliberate, and a reason to store it in a password manager as well as on the server.
Metadata stays readable: session ids, timestamps, roles, and the length of each message (a 120-character token above means roughly 24 characters of text). Fernet tokens also carry a plain timestamp.
The old text lingers unless you rewrite the file. My first test of "switch encryption on for an existing database" failed: after encrypting every row, the old plain text was still inside the file, in pages SQLite had freed but not erased. The fix is a VACUUM, which rewrites the file; the demonstration above shows Leeds gone afterwards. The same applies to rotating keys: without it, text encrypted under a retired key can remain in the file.
Old backups keep the old key's text. After a rotation, a backup made before it still needs the old key (shown above). Decide how long you keep both, and delete old backups.
It does not hide anything from the running bot, a root user on the server, or the model provider, who receive the conversation in the clear (chapters 2 and 6).
Whether this satisfies Telegram's terms, or the GDPR's security duties, is not something I can tell you. It is a reasonable technical measure; the legal judgement is yours.
Assembling the Service
build_service wires everything together from the settings, and its lifespan function owns start-up and shutdown. The shutdown order matters: signal the background tasks to stop and cancel them, give in-flight replies a bounded time to finish (drain, from chapters 9 and 10), close the HTTP clients to the platforms, and only then close the database. Closing the database first would turn every reply still in flight into an error.
@app.get("/readyz")
async def ready():
try:
store.db.execute("SELECT 1").fetchone()
except Exception: # noqa: BLE001 - readiness is a yes/no answer
from fastapi.responses import JSONResponse
return JSONResponse({"ready": False, "reason": "database"}, status_code=503)
return {"ready": True, "channels": sorted(k for k in channels if not k.endswith("_gateway"))}
async def purge_loop():
while True:
await asyncio.sleep(purge_every)
counts = await bot.purge_expired()
logger.info("retention_purge", extra=counts)
async def alert_loop():
last: set[str] = set()
while True:
await asyncio.sleep(alerts_every)
now = set(check_alerts(metrics))
for alert in now - last: # say it once when it starts, not every minute
logger.warning("alert", extra={"alert": alert})
last = now
@contextlib.asynccontextmanager
async def lifespan(_app):
stop = asyncio.Event()
tasks = [asyncio.create_task(factory(stop)) for factory in background]
tasks += [asyncio.create_task(purge_loop()), asyncio.create_task(alert_loop())]
logger.info("started", extra={"settings": s.summary()})
try:
yield
finally:
logger.info("stopping")
stop.set()
gateway = channels.get("discord_gateway")
if gateway:
gateway.stop()
for task in tasks:
task.cancel()
await asyncio.gather(*tasks, return_exceptions=True)
for name in ("telegram", "discord", "slack"): # let in-flight replies finish, but not forever
if name in channels:
with contextlib.suppress(asyncio.TimeoutError):
await asyncio.wait_for(channels[name][1].drain(), 10)
for name in ("telegram", "discord", "slack"):
if name in channels:
await channels[name][0].aclose()
store.close()
logger.info("stopped")
Two endpoints serve the platform around the bot. /healthz says the process is up and serving (nothing more, so a slow database cannot make an orchestrator kill a healthy process); /readyz also touches the database and returns 503 if it fails. Use the first for "restart me" and the second for "do not send me traffic".
The run command adds proxy_headers for the one address allowed to say who the client was (the local reverse proxy, never *), turns the per-request access log off (so message paths and addresses do not pile up in a second log), and allows 20 seconds for open connections to finish on shutdown.
Shutdown, Demonstrated
ready while running: {'ready': True, 'channels': []}
after the with-block (shutdown ran):
database closed: Cannot operate on a closed database.
log: started
log: stopping
log: stopped
This runs the same lifespan code in-process (it is Exercise 3). It shows the order of events, not a real signal: I did not send a real SIGTERM to a real process, and on Windows that is not a fair test of what systemd does on Linux. Test it on the server: systemctl stop chatbot while a streamed reply is running, and read the log for stopping and stopped.
Backups
Copying a SQLite file while the bot is writing to it can give you a torn copy that will not open. SQLite's online backup interface takes a consistent snapshot of a live database; Python exposes it as Connection.backup. The script opens the source read-only (a backup must never modify the live database; a test checks it does not even create one by mistake), checks the copy with PRAGMA integrity_check, keeps the newest N, and sets the files private.
"""Consistent backups of the SQLite database.
py -m bot.backup /var/lib/chatbot/chatbot.db /var/backups/chatbot --keep 7
Copying a SQLite file while the bot is writing to it can produce a torn, unusable copy. SQLite's online backup API takes a consistent
snapshot of a live database. Because message text is encrypted inside the database (when a content key is configured), the backup file
is encrypted too: keep the KEY somewhere else, or the backup protects nothing.
"""
import argparse
import pathlib
import sqlite3
import sys
import time
def backup_database(source: str, dest_dir: str, *, keep: int = 7, now: float | None = None) -> pathlib.Path:
"""Write a timestamped copy of the database to dest_dir, check it, and delete all but the newest `keep` copies."""
dest = pathlib.Path(dest_dir)
dest.mkdir(parents=True, exist_ok=True)
stamp = time.strftime("%Y%m%d-%H%M%S", time.gmtime(now if now is not None else time.time()))
target = dest / f"chatbot-{stamp}.db"
src = sqlite3.connect(f"file:{source}?mode=ro", uri=True) # read-only: a backup must never modify the live database
try:
out = sqlite3.connect(target)
try:
src.backup(out)
finally:
out.close()
finally:
src.close()
check = sqlite3.connect(target)
try:
result = check.execute("PRAGMA integrity_check").fetchone()[0]
finally:
check.close()
if result != "ok":
target.unlink()
raise RuntimeError(f"the backup failed its integrity check: {result}")
target.chmod(0o600)
for old in sorted(dest.glob("chatbot-*.db"))[:-keep] if keep > 0 else []:
old.unlink()
return target
def main(argv: list[str]) -> int:
parser = argparse.ArgumentParser(description="Back up the chatbot database.")
parser.add_argument("source")
parser.add_argument("dest_dir")
parser.add_argument("--keep", type=int, default=7)
args = parser.parse_args(argv)
try:
print(backup_database(args.source, args.dest_dir, keep=args.keep))
except (sqlite3.Error, RuntimeError, OSError) as e:
print(f"backup failed: {e}", file=sys.stderr)
return 1
return 0
if __name__ == "__main__":
sys.exit(main(sys.argv[1:]))
Three decisions that are yours. Where the copies go: the same disk is not a backup against a dead disk; copy them to another machine or storage. How long: a backup can hold conversations the bot has already deleted, so keep backups no longer than the retention period you told users (the seven daily copies here fit inside 30 days); and be honest that a /deleteme request does not reach copies already made, which is something to say in the privacy notice.
Whether it works: Exercise 2 is a restore drill. A backup you have never restored is a hope, not a backup.
Docker
Not run here (the engine was not running); written from Docker's documented instructions. A container gives you the same environment everywhere and an easy way to limit what the bot can touch. The choices in this file are the ones that matter: install from a pinned list (the versions installed in this project), run as an ordinary user, keep data in a volume so it survives the container, and give the platform a health check that needs no extra tools (Python is already there).
I did not verify that the base image tag exists; check it on Docker Hub, and rebuild regularly to receive security fixes.
FROM python:3.14-slim
ENV PYTHONUNBUFFERED=1 PYTHONDONTWRITEBYTECODE=1 CHATBOT_ENV=production CHATBOT_DATA_DIR=/data CHATBOT_HOST=0.0.0.0
WORKDIR /app
COPY deploy/requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY bot ./bot
COPY persona.toml .
# run as an ordinary user, never as root
RUN useradd --system --uid 10001 --no-create-home chatbot && mkdir /data && chown chatbot /data
USER chatbot
VOLUME /data
EXPOSE 8000
HEALTHCHECK --interval=30s --timeout=5s --start-period=10s --retries=3 \
CMD python -c "import urllib.request,sys; sys.exit(0 if urllib.request.urlopen('http://127.0.0.1:8000/healthz', timeout=3).status == 200 else 1)"
CMD ["python", "-m", "bot.app"]
127.0.0.1:8000:8000 publishes the port only to the machine itself, so the reverse proxy below is the only way in.
systemd on Debian
Without Docker, systemd starts the bot at boot, restarts it on failure and stops it on request (the manager itself is covered in systemd in Depth). Run the bot as its own account (useradd --system --home /opt/chatbot --shell /usr/sbin/nologin chatbot), install the code in /opt/chatbot with a virtual environment, and keep the environment file readable by root only.
The hardening directives limit the damage if the bot is ever compromised: no new privileges, a read-only file system except its own state directory, no access to home directories, a private /tmp, and no capabilities. The line ExecStartPre runs the configuration check first, so a bad configuration stops the start with the messages from earlier in this chapter in the journal.
$ systemd-analyze verify ./chatbot.service ./chatbot-backup.service ./chatbot-backup.timer (systemd 252, Debian 12 under WSL, copies in /tmp)
Configuration file /tmp/units/chatbot.service is marked executable. Please remove executable permission bits. Proceeding anyway.
chatbot.service: Command /opt/chatbot/venv/bin/python is not executable: No such file or directory
chatbot-backup.service: Command /opt/chatbot/venv/bin/python is not executable: No such file or directory
(the executable-bit lines are an artefact of copying from the Windows drive; the "No such file" lines are expected, because nothing is installed at /opt/chatbot here. No unknown directives or syntax errors were reported.)
That verification is narrow. It says the files parse, not that the sandbox lets the bot work. The usual surprise with ProtectSystem=strict is a write the bot did not know it was making: the backup unit needed its destination named in ReadWritePaths, which I added deliberately. After installing, run systemd-analyze security chatbot to see how exposed it is, and journalctl -u chatbot to see what it logged. The environment file template:
# Copy to /etc/chatbot/chatbot.env, chmod 600, owner root. Never commit the real file.
CHATBOT_ENV=production
CHATBOT_PROVIDER=claude
CHATBOT_MODEL=<read the current model name from your provider's model list>
ANTHROPIC_API_KEY=<your key>
# generate with: python -m bot.crypto_keygen (several keys, comma separated, NEWEST FIRST)
CHATBOT_CONTENT_KEYS=<fernet key>
# generate with: python -c "import secrets; print(secrets.token_urlsafe(48))"
CHATBOT_WEB_SECRET=<48+ random characters>
CHATBOT_LOG_KEY=<24+ random characters>
CHATBOT_ALLOWED_ORIGINS=https://chat.example.org
CHATBOT_RETENTION_DAYS=30
CHATBOT_DATA_DIR=/var/lib/chatbot
CHATBOT_PERSONA_FILE=/opt/chatbot/persona.toml
# optional channels
# TELEGRAM_BOT_TOKEN=
# TELEGRAM_WEBHOOK_SECRET=
# DISCORD_BOT_TOKEN=
# SLACK_BOT_TOKEN=
# SLACK_SIGNING_SECRET=
Fill in CHATBOT_MODEL yourself from your provider's current list: model names change, and a name I wrote here would be a guess. Create it with install -m 600 -o root so it is never world-readable, even for a moment.
HTTPS and the Reverse Proxy
The bot speaks plain HTTP on the local machine; a reverse proxy faces the internet, holds the certificate, and forwards. This is necessary, not decoration: Telegram's webhook needs HTTPS (chapter 9), Slack's request URL needs HTTPS (chapter 10), and the session cookie is marked Secure so browsers send it only over HTTPS. Not run here: this nginx file was not tested, because nginx is not installed and I did not install it; run nginx -t before reloading. Certificates from Let's Encrypt (certbot) are the usual free route; I did not obtain one.
Streaming. A proxy that buffers responses holds back a streamed reply until it is complete, so the typing effect from chapter 8 disappears. proxy_buffering off on the chat location is the fix, with a long proxy_read_timeout. The smoke test below prints a warning if a reply arrives as a single event.
Client address. The proxy sets the forwarding headers from the real connection ($remote_addr), never by passing along what the client claimed, and the bot trusts them only from 127.0.0.1.
Limits.limit_req rate-limits per address before a request even reaches the bot (a second layer beside chapter 7's limits), and client_max_body_size rejects oversized bodies early.
The firewall should allow 80 and 443 only. The bot's port is not published.
A Smoke Test
The last layer of chapter 11's list: does the deployed bot answer at all? The script checks both health endpoints, loads the page (which sets the cookie), and sends one real chat message the way a browser does. Here it ran against the real service started with py -m bot.app in production mode with the echo provider (a provider that repeats what you say, so no key was needed), on 127.0.0.1 without a proxy. That proves the assembled service works; it proves nothing about your server, the proxy, TLS or a real model.
"""A live smoke test: does the DEPLOYED bot answer at all? py deploy/smoke_test.py https://chat.example.org
Checks the health endpoints, loads the page (which sets the session cookie), and sends one chat message through the same path a
browser uses. It proves the wiring (proxy, TLS, service, provider) works; it says nothing about answer quality. One real message
costs a few tokens on a real provider.
"""
import json
import sys
import httpx
def main(base: str) -> int:
base = base.rstrip("/")
failures = []
with httpx.Client(base_url=base, timeout=60, headers={"Origin": base}) as client:
for path in ("/healthz", "/readyz"):
r = client.get(path)
print(f"{path:<10} {r.status_code} {r.text.strip()[:80]}")
if r.status_code != 200:
failures.append(path)
page = client.get("/")
print(f"{'/':<10} {page.status_code} CSP: {'yes' if 'content-security-policy' in page.headers else 'MISSING'} cookie: {'yes' if client.cookies else 'MISSING'}")
if page.status_code != 200 or not client.cookies:
failures.append("page/cookie")
pieces = 0
text = ""
with client.stream("POST", "/api/chat", json={"message": "Say hello in five words."}) as r:
print(f"/api/chat {r.status_code} content-type: {r.headers.get('content-type')}")
for line in r.iter_lines():
if line.startswith("data: "):
event = json.loads(line[6:])
pieces += 1
text = event.get("text", text)
print(f" {pieces} streamed event(s); final reply: {text[:80]!r}")
if r.status_code != 200 or not text:
failures.append("chat")
if pieces < 2:
print(" WARNING: the reply arrived as a single event; a proxy may be buffering the stream")
print("SMOKE TEST", "FAILED: " + ", ".join(failures) if failures else "PASSED")
return 1 if failures else 0
if __name__ == "__main__":
sys.exit(main(sys.argv[1] if len(sys.argv) > 1 else "http://127.0.0.1:8000"))
/healthz 200 {"status":"ok"}
/readyz 200 {"ready":true,"channels":[]}
/ 200 CSP: yes cookie: yes
/api/chat 200 content-type: text/event-stream; charset=utf-8
2 streamed event(s); final reply: "I'm an AI assistant, not a person. Type /privacy to see what I store.\n\nYou said:"
SMOKE TEST PASSED
After a deployment, run it against the public address. It costs a few tokens on a real provider; run it after each deployment, not every minute.
The Runbook
A short document for the day something is wrong, written while you are calm:
Situation
First look
Usual cause and action
The bot will not start
journalctl -u chatbot -n 50
The configuration check printed its problems; fix the environment file. A WrongKey means the content key does not match the database: restore the right key, never "start fresh" over the data.
Users get errors
The log: provider_call lines with error and status
Provider outage or key or quota problem. The breaker and fallback from chapter 7 should be answering; check the provider's status page.
Replies arrive all at once
The smoke test warning
A proxy buffering the stream; see the nginx section.
Slack or Telegram stopped delivering
The platform's own delivery view; /readyz
Certificate expired, the signing secret changed, or the webhook URL changed. Renew or reset and run --set-webhook.
Cost spike
alert lines; token totals
The daily budget (chapter 7) limits the damage; find the user (by pseudonym) and decide.
Database damaged or deleted
PRAGMA integrity_check
Stop the service, restore the newest good backup (Exercise 2), start, run the smoke test.
A key or token leaked
When and where
Revoke the platform token first and issue a new one; rotate the content key (put the new key first, run rotate, then retire the old one). If personal data may have been exposed, the breach duties in your jurisdiction apply: that is a decision for you and, probably, an adviser.
A user asks for their data
The commands from chapter 6
/privacy, the export, /deleteme. Remember copies in backups and pseudonymised log lines expire on their own schedule.
Mistakes while building this
"Encrypted" rows, plain bytes. Described above: the old plain text survived inside the file until a VACUUM. What went wrong: I assumed that rewriting each value replaced it on disk. How to avoid it: test security properties against the raw bytes of the file, not against what the API returns; the test that looked at the file caught it.
A summary that told the truth about the wrong thing. The production summary showed development defaults as "set". Avoid it by deriving what you print from the same values the program will use, and by reading the output, not just the exit code.
A test that left a file open. One service test built a service it never started, so nothing closed its database, and on Windows the temporary directory could not be deleted. Takeaway: whatever opens a resource in a test also closes it, and the failure was a useful reminder that on Linux the same leak would be silent.
Killing the wrong thing. After the smoke test I looked for the process I had started and first used the process id the shell reported, which is not the Windows one, so the stop failed with "cannot find a process". I then matched it by its start time. Takeaway: identify a process by something you can verify (here, its start time and path) before stopping it, and never stop processes by name.
Not run here: no keys or local model were available; the code follows each vendor's documentation as I read it in the earlier chapters
Telegram, Discord and Slack against the real platforms
Not run here: simulated with local servers and recorded-shape messages
Legal compliance (UK GDPR, platform terms)
Technical measures only; not legal advice
Hands-On Exercises
Exercise 1
Write a deliberately broken production environment (a bad port, a bad content key, a short web secret, Slack half-configured, a real provider with no key) and show every problem reported in one pass. Then fix them in groups and show the count fall to zero.
Run a restore drill: fill an encrypted database, back it up, delete the live file, restore from the backup, check integrity and message count, read a message with the right key, and show what the wrong key does.
Prove the shutdown sequence: run the assembled service in-process, send it a request, leave the block, then show that the database is closed and the log contains started, stopping and stopped in order. Say what this does not prove.
Configuration from environment variables; validate at start-up, report all problems, refuse insecure production settings, never print secrets (py -m bot.app --check)
Encrypt message text in the application (Fernet, MultiFernet for rotation); key kept apart from data and backups; metadata and lengths stay visible; VACUUM after rewriting; test against the raw file
/healthz = process alive (restart me); /readyz = database reachable (send me traffic)
Shutdown order: stop tasks, drain in-flight replies (bounded), close clients, close the database last
Backups: SQLite online backup, integrity check, keep N, shorter than your retention promise, stored elsewhere, restore drill
Docker: pinned dependencies, non-root, volume for data, health check, bind the port to localhost
systemd: own user, EnvironmentFile (mode 600), ExecStartPre config check, hardening directives, restart on failure, timer for backups
Proxy: TLS, forwarding headers, proxy_buffering off for streams, rate and body limits, firewall to 80/443 only
After every deploy: smoke test. Keep a runbook. Label what you have not tested
Course Complete: What You Can Now Do
Chapter
You can now…
1 Anatomy of a Bot
Explain the parts a bot adds to a model call and how a message travels through them
2 Provider Interface
Put Claude and OpenAI-compatible models (Grok, OpenAI, Ollama) behind one interface with typed errors and streaming
3 Local Models
Reason about running a model locally and what it costs you
4 Memory
Keep history, a token budget, summaries and user facts, safely
5 Persona and Guardrails
Write a persona and test that it holds, with input and output guards
6 Moderation and Privacy
Moderate, handle crisis messages, honour privacy requests and retention
7 Cost and Reliability
Limit usage and spending; retry, break and fall back
8 Web Chat
Serve a streaming chat page with sessions and CSRF protection
9 Telegram
Run the same core on Telegram, with polling or webhook
10 Discord and Slack
Add two more channels without touching the core
11 Test and Observe
Log without leaking, measure, alert, and test with golden conversations and simulations
12 Deploy
Configure, encrypt, back up, containerise, run as a service, put it behind HTTPS, and say honestly what you have verified
Where to go next: run chapter 12's checks on your own server and fix what the "not run here" items turn up; try the real-model scenarios from chapter 5 with a real key and look at the failure rate; and read the evaluation methods in Advanced AI Workflows & the Claude API 9, since nothing in this course says whether the bot's answers are any good. For the service layer, systemd in Depth and the Docker and web-server courses cover what this chapter only used.