Written with AI assistance from the source video. The video is the authority.
Log in to leave a comment
No posts yet
The AI automation you see on YouTube is easy and flashy. It feels like everything is done once you run the code on your local computer and it works. But that's only halfway there.
The real problems start the moment you upload that script to a cloud server. You wake up in the morning to find the bot that was running fine until yesterday is dead. It's either silently stopped without an error message, or in the worst-case scenario, it's stuck in an infinite loop, billing you hundreds of dollars in API costs.
Subtle environmental differences between local and server, API calls without exception handling, and a lack of monitoring are barriers that junior developers and solopreneur knowledge founders inevitably face. I have organized the 3 foundational pillars you need to overhaul right now to build an automation system that runs stably 24/7.
The first reason Python-based automation systems stop on a server is library version mismatch. Many people use the pip freeze > requirements.txt method. However, this method has a fatal flaw where sub-package versions change arbitrarily depending on the environment or operating system at the time of installation. This is why things work on your computer but don't even run on the server due to package conflicts.
In fact, the engineering team at Rippling, a global HR management platform company in the U.S., boldly ditched their existing package management tools and adopted uv, a Rust-based package manager, to solve dependency hell. This is a case where productivity was increased by reducing build and library installation time from 10 minutes to less than 1 minute.
To ensure your automation doesn't die, you must completely isolate the virtual environment and strictly lock even sub-package versions. You can solve this problem by using Poetry, which supports the Python standard pyproject.toml.
First, open your terminal, install Poetry, and create a project folder.
bash curl -sSL https://install.python-poetry.org | python3 - poetry new ai_automation_project cd ai_automation_project
Next, change the setting so that the virtual environment is created inside the project folder (-project) rather than in a system-wide folder. This ensures that things don't get tangled when you upload the entire folder to the server.
bash poetry config virtualenvs.in-project true --local
Now, add the necessary external libraries and development test tools. A poetry.lock file containing unique hash values is created, and the environment is completely fixed.
bash poetry add anthropic python-dotenv sqlite3 poetry add --group dev pytest pytest-mock poetry install --sync
If a .venv directory has been created in your project root folder, you're successful. Simply syncing library versions between local and server can cut the time wasted on deployment malfunctions by more than half.
Manually executing a script locally and having it succeed is not validation. A bot deployed to a cloud server is defenselessly exposed to all kinds of error environments, such as network latency, temporary API server outages, or unexpected empty string returns. If proper exception handling isn't in place when a temporary communication error occurs, the bot will die on the spot or push broken data directly into the database.
To perfectly validate these exception scenarios while saving on paid API call costs, you should write unit tests that utilize mock objects (Mock) instead of sending requests to the actual API server.
First, isolate the external API call class using a dependency injection pattern and write app_bot.py.
`python
import time
import anthropic
from anthropic import Anthropic
class ClaudeAutomationBot:
def init(self, client: Anthropic = None):
self.client = client or Anthropic()
def process_task(self, prompt: str, timeout_limit: float = 5.0) -> dict:
if not prompt.strip():
raise ValueError("The input prompt is not valid.")
start_time = time.perf_counter()
try:
response = self.client.with_options(timeout=timeout_limit).messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=1000,
messages=[{"role": "user", "content": prompt}]
)
elapsed_time = time.perf_counter() - start_time
if not response.content or not response.content[0].text:
raise ValueError("No text data block in API response.")
return {
"status": "success",
"text": response.content[0].text,
"latency": elapsed_time
}
except anthropic.APITransitError:
return {"status": "error", "reason": "timeout", "latency": time.perf_counter() - start_time}
except Exception as e:
return {"status": "error", "reason": str(e), "latency": time.perf_counter() - start_time}
`
Now, create test_app_bot.py in the same directory to inject fake responses and forcibly simulate normal operations and abnormal data.
`python
import pytest
from app_bot import ClaudeAutomationBot
from anthropic import Anthropic
from anthropic.types import Message, Usage
from anthropic.types.content_block import ContentBlock
def test_api_response_data_presence_success(mocker):
mock_client = mocker.MagicMock(spec=Anthropic)
mock_message = mocker.MagicMock(spec=Message)
mock_content = mocker.MagicMock(spec=ContentBlock)
mock_content.type = "text"
mock_content.text = "This is data with perfect processing and integrity."
mock_message.content = [mock_content]
mock_message.usage = Usage(input_tokens=15, output_tokens=25)
mock_client.messages.create.return_value = mock_message
bot = ClaudeAutomationBot(client=mock_client)
result = bot.process_task("Normal summary test")
assert result["status"] == "success"
assert result["text"] == "This is data with perfect processing and integrity."
assert result["latency"] > 0
def test_api_response_empty_data_failure(mocker):
mock_client = mocker.MagicMock(spec=Anthropic)
mock_message = mocker.MagicMock(spec=Message)
mock_content = mocker.MagicMock(spec=ContentBlock)
mock_content.type = "text"
mock_content.text = ""
mock_message.content = [mock_content]
mock_client.messages.create.return_value = mock_message
bot = ClaudeAutomationBot(client=mock_client)
result = bot.process_task("Abnormal input test")
assert result["status"] == "error"
assert "No text data block" in result["reason"]
`
Finally, build a CI pipeline that automatically runs these tests whenever code is pushed to the .github/workflows/python-package.yml path in your GitHub repository.
`yaml
name: Python Automation Bot Test and Deploy
on:
push:
branches: [ main ]
pull_request:
branches: [ main ]
jobs:
test:
runs-on: ubuntu-latest
steps:
- name: Checkout repository source
uses: actions/checkout@v4
- name: Set up Python 3.12
uses: actions/setup-python@v5
with:
python-version: "3.12"
- name: Install Poetry and set up environment
uses: abatilo/actions-poetry@v4
- name: Install project dependencies
run: |
poetry config virtualenvs.create true --local
poetry config virtualenvs.in-project true --local
poetry install --no-interaction
- name: Run unit tests for integrity verification
run: |
poetry run pytest -v
`
Linking this test in a cloud environment like Railway or Vercel will completely block deployment if even a single test fails. It prevents disasters where the entire production server is paralyzed due to a minor typo or incorrect exception handling.
The scariest reality faced when operating an automation infrastructure is cost. If loop termination conditions are messed up and it falls into an infinite loop, or if prompt caching settings are disabled and heavy context continues to be transmitted, you will receive a bill for hundreds of dollars when you wake up.
To prevent this, you need a hardware-level kill switch that checks the local DB before sending an API call to calculate the accumulated daily spending and forcibly stops the system if it exceeds the daily limit you've set.
First, create database_manager.py using SQLite to record API call usage.
`python
import sqlite3
import time
class DBUsageLogger:
def init(self, db_path: str = "monitoring_usage.db"):
self.db_path = db_path
self._setup_table()
def _setup_table(self):
with sqlite3.connect(self.db_path) as conn:
conn.execute("""
CREATE TABLE IF NOT EXISTS api_metrics (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp REAL,
model TEXT,
input_tokens INTEGER,
cache_read_tokens INTEGER,
output_tokens INTEGER,
cost_usd REAL
)
""")
conn.commit()
def get_today_accumulated_spend(self) -> float:
utc_today_start = time.time() - (time.time() % 86400)
with sqlite3.connect(self.db_path) as conn:
cursor = conn.cursor()
cursor.execute(
"SELECT SUM(cost_usd) FROM api_metrics WHERE timestamp >= ?",
(utc_today_start,)
)
result = cursor.fetchone()
return result[0] if result[0] is not None else 0.0
def insert_transaction(self, model: str, input_t: int, cache_r_t: int, output_t: int, cost: float):
with sqlite3.connect(self.db_path) as conn:
conn.execute("""
INSERT INTO api_metrics
(timestamp, model, input_tokens, cache_read_tokens, output_tokens, cost_usd)
VALUES (?, ?, ?, ?, ?, ?)
", (time.time(), model, input_t, cache_r_t, output_t, cost))
conn.commit()
`
Next, write kill_switch_middleware.py, a middleware that checks the budget before an API call and calculates costs in real-time based on a rate table. The rates are based on official Anthropic pricing ($3 per 1M input tokens, $15 for output, $0.3 for prompt cache read).
`python
from database_manager import DBUsageLogger
class APIBudgetKillSwitch:
def init(self, logger_db: DBUsageLogger, max_daily_budget_usd: float = 5.00):
self.db = logger_db
self.limit = max_daily_budget_usd
def check_pre_flight_safety(self):
today_spend = self.db.get_today_accumulated_spend()
if today_spend >= self.limit:
raise PermissionError(
f"[SECURITY BLOCK] Daily API usage limit exceeded! "
f"Set budget: ${self.limit:.2f}, Current total: ${today_spend:.4f}. "
f"Forcing process shutdown for safety."
)
def calculate_and_save_cost(self, model: str, input_t: int, cache_r_t: int, output_t: int) -> float:
pricing_chart = {
"claude-3-5-sonnet-20241022": {"input": 3.0, "cache_read": 0.3, "output": 15.0},
"claude-3-5-haiku-20241022": {"input": 1.0, "cache_read": 0.1, "output": 5.0},
}
prices = pricing_chart.get(model, {"input": 3.0, "cache_read": 0.3, "output": 15.0})
fresh_input = input_t - cache_r_t
cost = (
(fresh_input * (prices["input"] / 1000000.0)) +
(cache_r_t * (prices["cache_read"] / 1000000.0)) +
(output_t * (prices["output"] / 1000000.0))
)
self.db.insert_transaction(model, input_t, cache_r_t, output_t, cost)
return cost
`
When the bot runs, it calls check_pre_flight_safety() at the very front of the loop to check for safety, and after the API call, it saves the number of used tokens and costs to the database in real-time.
If more than $5 is spent in a day, the program cuts off communication and shuts down itself. It's a hundred times safer to have the script shut down on its own than to have your wallet emptied by a malfunction.
Even after safely deploying to the cloud, it can stop due to unexpected external server errors. It's a waste of time for a junior developer or solopreneur to stare at the real-time log window of a cloud dashboard all day.
You need a system that immediately sends a Telegram notification when an error stops the program, detailing exactly which line it crashed on and what the variable values were.
`python
import html
import json
import logging
import traceback
import requests
from datetime import datetime
logger = logging.getLogger("ProdAppLogger")
logger.setLevel(logging.INFO)
log_format = logging.Formatter(
"[%(asctime)s] [%(levelname)s] [%(filename)s:%(lineno)d]: %(message)s"
)
console_handler = logging.StreamHandler()
console_handler.setFormatter(log_format)
logger.addHandler(console_handler)
class TelegramEmergencyNotifier:
def init(self, bot_token: str, target_chat_id: str):
self.token = bot_token
self.chat_id = target_chat_id
self.api_url = f"https://api.telegram.org/bot{self.token}/sendMessage"
def dispatch_error_log(self, error: Exception, context_data: dict = None):
trace_back = traceback.format_exception(type(error), error, error.__traceback__)
joined_trace = "".join(trace_back)
safe_trace = html.escape(joined_trace)
safe_context = html.escape(json.dumps(context_data or {}, indent=2, ensure_ascii=False))
if len(safe_trace) > 3000:
safe_trace = safe_trace[:3000] + "\n[...log truncated below...]"
alert_body = (
f"⚠️ <b>[Automation Bot Operation Emergency Error Alert]</b>\n"
f"📅 <b>Occurrence Time (KST):</b> {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n"
f"💻 <b>Module:</b> AI Automation Pipeline\n\n"
f"⚙️ <b>Context state data before/after error:</b>\n"
f"<pre>{safe_context}</pre>\n\n"
f"📁 <b>Detailed Traceback:</b>\n"
f"<pre>{safe_trace}</pre>"
)
payload = {
"chat_id": self.chat_id,
"text": alert_body,
"parse_mode": "HTML"
}
try:
response = requests.post(self.api_url, json=payload, timeout=10)
if response.status_code != 200:
logger.error(f"Telegram Alert failed. Status: {response.status_code}")
except requests.exceptions.RequestException as net_err:
logger.error(f"Alert delivery network crash. Details: {str(net_err)}")
`
Connect this alert module to the outermost try-except block wrapping the main process. When an error occurs, the source code file location and exception message are sent to your mobile device in real-time based on the error traceback.
You don't have to spend a long time searching through unfamiliar English logs in the server terminal, as you can see the notification and start repair work on the spot, speeding up your response time.
Anyone can write code that works once or twice. But keeping it running flawlessly on a server 365 days a year is a completely different story.
Start by blocking global virtual environment pollution and aligning dependencies with Poetry. Embed a rate table in a local DB to protect your wallet, and get real-time situational awareness via Telegram when an error occurs. Only when these infrastructure-level mechanisms are layered on top of each other will your AI automation bot truly act as a reliable worker that helps your business.