Chapter 16: Python Services

Last Updated: 2026-03

16.2 Flask Pattern — Patriot Agent

The Patriot Agent is a Flask web service that aggregates news, analyzes it with Claude, and delivers reports on a schedule.

Project structure

patriot-agent/
├── app.py                  # Flask application factory
├── scheduler.py            # APScheduler job definitions
├── services/
│   ├── claude_service.py   # Anthropic Claude integration
│   ├── tavily_service.py   # Tavily web search
│   └── rss_service.py      # feedparser RSS ingestion
├── models/
│   └── report.py           # Data models
├── templates/              # Jinja2 HTML templates
├── requirements.txt        # Dependencies
├── .env                    # Secrets (not committed)
└── CLAUDE.md

Application factory pattern

# app.py
from flask import Flask
from .scheduler import init_scheduler

def create_app() -> Flask:
    app = Flask(__name__)
    app.config.from_object('config.ProductionConfig')

    # Initialize services
    init_scheduler(app)

    # Register blueprints
    from .routes import main_bp
    app.register_blueprint(main_bp)

    return app

APScheduler integration

APScheduler runs background jobs on a cron schedule within the Flask process:

# scheduler.py
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger

def init_scheduler(app: Flask) -> None:
    scheduler = BackgroundScheduler()

    scheduler.add_job(
        func=run_daily_report,
        trigger=CronTrigger(hour=7, minute=0),  # 7:00 AM daily
        id='daily_report',
        replace_existing=True,
    )

    scheduler.start()

Claude integration

# services/claude_service.py
import anthropic
import os

class ClaudeService:
    def __init__(self):
        self._client = anthropic.Anthropic(api_key=os.environ['ANTHROPIC_API_KEY'])

    def analyze(self, content: str, system_prompt: str) -> str:
        message = self._client.messages.create(
            model="claude-opus-4-5",
            max_tokens=4096,
            system=system_prompt,
            messages=[{"role": "user", "content": content}]
        )
        return message.content[0].text

Tavily search integration

# services/tavily_service.py
from tavily import TavilyClient
import os

class TavilyService:
    def __init__(self):
        self._client = TavilyClient(api_key=os.environ['TAVILY_API_KEY'])

    def search(self, query: str, max_results: int = 10) -> list[dict]:
        response = self._client.search(query=query, max_results=max_results)
        return response.get('results', [])

---

16.3 FastAPI Pattern

FastAPI is the pattern for new Python API services that require:

• High performance (async)
• Automatic OpenAPI documentation
• Pydantic input validation

Project structure

my-fastapi-service/
├── main.py                 # FastAPI app + router registration
├── routers/
│   ├── ai.py               # AI-related endpoints
│   └── health.py           # Health check endpoint
├── schemas/
│   └── requests.py         # Pydantic request/response models
├── services/
│   └── claude_service.py   # AI service layer
├── dependencies.py         # FastAPI dependency injection
├── requirements.txt
└── CLAUDE.md

App setup

# main.py
from fastapi import FastAPI
from .routers import ai, health

app = FastAPI(
    title="ITI AI Service",
    version="1.0.0",
    docs_url="/docs",  # Disable in production if not needed
)

app.include_router(health.router)
app.include_router(ai.router, prefix="/api/v1")

Pydantic models

# schemas/requests.py
from pydantic import BaseModel, Field

class AIRequest(BaseModel):
    user_message: str = Field(..., min_length=1, max_length=10000)
    context: str | None = None
    max_tokens: int = Field(default=2048, ge=1, le=8192)

class AIResponse(BaseModel):
    content: str
    model: str
    usage: dict

Async endpoint

# routers/ai.py
from fastapi import APIRouter, Depends, HTTPException
from ..schemas.requests import AIRequest, AIResponse
from ..services.claude_service import ClaudeService

router = APIRouter()

async def get_claude_service() -> ClaudeService:
    return ClaudeService()

@router.post("/complete", response_model=AIResponse)
async def complete(
    request: AIRequest,
    claude: ClaudeService = Depends(get_claude_service),
) -> AIResponse:
    try:
        result = await claude.complete(request.user_message, request.context)
        return AIResponse(**result)
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

Running with Uvicorn

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

---

16.4 Environment Variables

All Python services read configuration from environment variables. Never hardcode secrets.

Standard pattern

import os
from dotenv import load_dotenv

load_dotenv()  # Load .env file for local development

# Validate required variables at startup
REQUIRED_ENV_VARS = ['ANTHROPIC_API_KEY', 'TAVILY_API_KEY']
missing = [v for v in REQUIRED_ENV_VARS if not os.environ.get(v)]
if missing:
    raise EnvironmentError(f"Missing required environment variables: {missing}")

ANTHROPIC_API_KEY = os.environ['ANTHROPIC_API_KEY']

---

16.5 Dependencies

Use requirements.txt for all Python services. Pin major versions:

anthropic>=0.40.0,<1.0
flask>=3.0.0,<4.0
flask-apscheduler>=1.13.0,<2.0
fastapi>=0.115.0,<1.0
uvicorn[standard]>=0.32.0,<1.0
pydantic>=2.0.0,<3.0
httpx>=0.27.0,<1.0
tavily-python>=0.5.0,<1.0
feedparser>=6.0.0,<7.0
python-dotenv>=1.0.0,<2.0

Install:

pip install -r requirements.txt

---

Previous: Chapter 15 — Desktop Apps with Tauri 2 | Next: Chapter 17 — iOS & macOS with Swift