AI Incident Triage Assistant
Build a FastAPI service that turns raw error logs into structured AI analysis.
Introduction
30 Second Summary
When an application crashes at 3 AM, the first minutes matter most. Engineers scramble to read stack traces, guess the severity, and figure out what to fix first.
In this project, you will build an AI-powered incident triage API using FastAPI and the Anthropic Claude API. You will send raw error logs to a single endpoint and get back structured JSON with severity levels, root causes, and recommended fixes.
What You'll Build
Picture curling your running API with a gnarly Java OutOfMemoryError stack trace and getting back clean, validated JSON that tells you the severity is critical, the likely cause is a memory leak, and your first fix should be to check heap allocation settings.
By the end of this project, you'll have:
- An AI-powered triage endpoint you can curl with any error log and get back consistent, validated JSON breaking down severity, category, root causes, and fix recommendations.
- Production observability and resilience so every request is traceable by a unique ID, logged as structured JSON, validated for size, and protected from upstream failures with proper timeout handling and error codes.
- A Docker-packaged service you can start on any machine with a single command, fully portable and deployment-ready.
- Secret Mission: Add deterministic escalation rules that override AI severity for critical patterns like OOM and segfaults.
Are there any prerequisites?
This project assumes you are comfortable with Python fundamentals and have basic familiarity with REST APIs and terminal commands. No prior experience with FastAPI or LLM integration is needed.
Before We Start
Before we dive in, let's lock in what you're about to build and why it matters.
Set Up the Project
Before you can build any triage logic, the project needs a foundation. That means a Python environment with the right packages, a secure way to load your API key, and a confirmed connection to Claude.
In this step, you'll create the project directory, install the dependencies, configure your API key with pydantic-settings, and run a quick test script to prove everything is wired up.
In this step, get ready to:
- Install FastAPI, the Anthropic Python SDK, uvicorn, and pydantic-settings.
- Configure the API key securely with a Settings class and a .env file.
- Verify the Claude API connection with a test script.
Create the project and install dependencies
- Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
- Type Terminal and press Enter to open it.
- Move to your Desktop by running this command:
cd ~/Desktop
- Create your project folder and move into it by running these commands:
mkdir incident-triage-assistant
cd incident-triage-assistant
- Confirm you're in the new folder by running:
ls
The folder is empty, so ls prints nothing. That's expected.
- Open the incident-triage-assistant folder in your code editor.
- Create a new file called requirements.txt and add the following content:
fastapi==0.141.1
uvicorn==0.52.4
anthropic==1.0.0
pydantic-settings==2.15.0
What are these dependencies?
- fastapi is the web framework for building the API endpoints.
- uvicorn is the ASGI server that runs the FastAPI app.
- anthropic is the official Python SDK for calling the Claude API.
- pydantic-settings loads configuration from environment variables and .env files into typed Python objects.
- Save requirements.txt.
- Install the dependencies by running this command in your terminal:
pip install -r requirements.txt
You should see pip downloading and installing each package. The output ends with a line listing all successfully installed packages.
Seeing a permissions error or command not found?
- If pip is not found, try pip3 instead. Some systems use pip3 for Python 3.
- If you see a permissions error, try adding --user at the end of the command.
Still stuck? Help me fix my pip install error.
Set up the API key and app configuration
The Anthropic SDK needs your API key to authenticate requests. Instead of hardcoding it, you'll store it in a .env file and load it with pydantic-settings.
- Create a new file called .env in the root of your incident-triage-assistant folder.
- Add this line, replacing your-api-key-here with your actual Anthropic API key:
ANTHROPIC_API_KEY=your-api-key-here
Why a .env file?
Storing secrets in a .env file keeps them out of your source code. You can add .env to .gitignore so your key never gets committed to version control.
- Create the app/ package directory and its __init__.py file by running these commands:
mkdir app
touch app/__init__.py
- Confirm the new structure by running:
ls
You should see app, .env, and requirements.txt listed.
- Create a new file called config.py inside the app/ folder and add the following code:
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
# Required: loaded from ANTHROPIC_API_KEY in .env
anthropic_api_key: str
# Claude model to use for triage analysis
model_name: str = "claude-haiku-4-5"
# Max tokens for Claude's response
max_tokens: int = 1024
# Timeout in seconds for Claude API calls
claude_timeout: float = 30.0
# Logging level
log_level: str = "INFO"
# Tell pydantic-settings to read from a .env file
model_config = {"env_file": ".env"}
What does this code do?
- Settings extends BaseSettings from pydantic-settings, which automatically maps environment variables to class fields.
- anthropic_api_key has no default, so pydantic-settings will read ANTHROPIC_API_KEY from your .env file. If the key is missing, the app raises an error immediately.
- model_config = {"env_file": ".env"} tells pydantic-settings where to find the environment file.
- Save app/config.py.
✔️ Awesome, I've got everything!
Great. Double check you've saved app/config.py and that your .env file has your real API key in it.
ⓧ I'd like to double check the full code
Here is the complete app/config.py:
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
anthropic_api_key: str
model_name: str = "claude-haiku-4-5"
max_tokens: int = 1024
claude_timeout: float = 30.0
log_level: str = "INFO"
model_config = {"env_file": ".env"}
Verify the API connection
You have the dependencies installed and the API key configured. Now it's time to prove they actually work together.
- Create a new file called test_api.py in the root of your incident-triage-assistant folder (next to requirements.txt, not inside app/) and add the following code:
from anthropic import Anthropic
from app.config import Settings
# Load settings from .env
settings = Settings()
# Create an Anthropic client with your API key
client = Anthropic(api_key=settings.anthropic_api_key)
# Send a simple test message to Claude
message = client.messages.create(
model=settings.model_name,
max_tokens=256,
messages=[{"role": "user", "content": "Say hello!"}],
)
# Print Claude's response
print(message.content[0].text)
What does this script do?
- Settings() reads your ANTHROPIC_API_KEY from the .env file and loads all the config defaults.
- Anthropic(api_key=...) creates a client that authenticates with the Claude API.
- messages.create() sends a request to claude-haiku-4-5 and returns Claude's response.
- message.content[0].text extracts the text from Claude's first content block.
- Save test_api.py.
Before you run this, what do you expect to see? Think about what the script does and what Claude might respond to "Say hello!"
- Run the test script by running this command:
python test_api.py
You should see a friendly greeting from Claude printed in your terminal. Something like "Hello! How can I help you today?" The exact wording varies, but any response means your API key, SDK, and configuration are all working.
Seeing an authentication error?
- Open your .env file and confirm ANTHROPIC_API_KEY is set to your actual key with no extra spaces or quotes around the value.
- Make sure the key starts with sk-ant-. If it looks different, double check it on the Anthropic console.
- If you see a ModuleNotFoundError, make sure you ran pip install -r requirements.txt successfully in the earlier substep.
Still stuck? Help me troubleshoot my Claude API test script.
✔️ Awesome, I've got everything!
Great. Make sure test_api.py is saved and you saw a response from Claude in the terminal.
ⓧ I'd like to double check the full code
Here is the complete test_api.py:
from anthropic import Anthropic
from app.config import Settings
settings = Settings()
client = Anthropic(api_key=settings.anthropic_api_key)
message = client.messages.create(
model=settings.model_name,
max_tokens=256,
messages=[{"role": "user", "content": "Say hello!"}],
)
print(message.content[0].text)
Your project is set up and Claude is responding. Next up, you'll use FastAPI to build a real API endpoint that sends error logs to Claude for analysis.
Build the Raw Triage Endpoint
Your Python project is ready and Claude answered your test script. Now it's time to put that API connection to work inside a real web service.
The core idea is straightforward. Accept an error log, send it to Claude, and return the analysis. In this step, you'll build a raw version of the triage endpoint that returns Claude's free-text response directly.
In this step, get ready to:
- Create app/main.py with a FastAPI app and a POST /triage endpoint.
- Test the endpoint with a sample error log and see Claude's analysis.
Create the FastAPI app and request model
Your triage API needs an entry point. The app/main.py file creates the FastAPI application and defines the shape of incoming requests using a Pydantic model.
- Create a new file called main.py inside the app/ folder of your project.
- Add the following code to set up the app and request model:
from anthropic import Anthropic
from fastapi import FastAPI
from pydantic import BaseModel
from app.config import Settings
# Load settings from .env via pydantic-settings
settings = Settings()
# Create the FastAPI application
app = FastAPI(title="AI Incident Triage Assistant", version="1.0.0")
# Define the expected shape of incoming triage requests
class TriageRequest(BaseModel):
log_text: str
source: str = "unknown"
What does this code do?
Settings() loads your ANTHROPIC_API_KEY from the .env file automatically. This is the same Settings class you tested in Step 1.
TriageRequest is a Pydantic model that tells FastAPI what shape incoming JSON must have. Every request needs a log_text string. The source field defaults to "unknown" if not provided.
- Save app/main.py.
Add the triage endpoint
The endpoint receives an error log, passes it to Claude with a prompt asking for triage analysis, and returns Claude's raw text response.
- In app/main.py, add the following code below the TriageRequest class:
@app.post("/triage")
async def triage_incident(payload: TriageRequest):
# Create an Anthropic client for this request
client = Anthropic(api_key=settings.anthropic_api_key)
# Send the log to Claude with a basic analysis prompt
message = client.messages.create(
model=settings.model_name,
max_tokens=settings.max_tokens,
messages=[
{
"role": "user",
"content": f"Analyze this error log and provide severity, category, summary, likely causes, and recommended actions:\n\nSource: {payload.source}\n\nLog:\n{payload.log_text}",
}
],
)
# Return Claude's raw text response wrapped in JSON
return {"analysis": message.content[0].text}
What does this endpoint do?
The @app.post("/triage") decorator registers a POST route at /triage. FastAPI automatically parses incoming JSON against the TriageRequest model before your function runs.
Inside the handler, a new Anthropic client sends the log text to claude-haiku-4-5 with a prompt asking for severity, category, summary, causes, and actions. The raw text response comes back inside a JSON wrapper.
- Save app/main.py.
✔️ Awesome, I've got everything!
Great. Make sure you've saved app/main.py before moving on.
ⓧ I'd like to double check the full code
Here is the complete app/main.py file. Make sure your code matches this exactly:
from anthropic import Anthropic
from fastapi import FastAPI
from pydantic import BaseModel
from app.config import Settings
settings = Settings()
app = FastAPI(title="AI Incident Triage Assistant", version="1.0.0")
class TriageRequest(BaseModel):
log_text: str
source: str = "unknown"
@app.post("/triage")
async def triage_incident(payload: TriageRequest):
client = Anthropic(api_key=settings.anthropic_api_key)
message = client.messages.create(
model=settings.model_name,
max_tokens=settings.max_tokens,
messages=[
{
"role": "user",
"content": f"Analyze this error log and provide severity, category, summary, likely causes, and recommended actions:\n\nSource: {payload.source}\n\nLog:\n{payload.log_text}",
}
],
)
return {"analysis": message.content[0].text}
Start the server and test the endpoint
With the code saved, you can start the API server and send it a real error log to triage.
- Start the uvicorn development server by running this command:
uvicorn app.main:app --reload
What does this command do?
uvicorn is a fast ASGI server that runs your FastAPI app. The app.main:app part tells it to find the app object inside app/main.py.
The --reload flag watches your files for changes and automatically restarts the server when you save. This is for development only.
You should see output showing the server is running at http://127.0.0.1:8000.
Server not starting?
- Make sure you're running the command from your project's root folder (the one containing requirements.txt), not from inside the app/ folder.
- Check that app/__init__.py exists. Python needs this file to treat app/ as a package.
- If you see a ModuleNotFoundError, confirm your virtual environment is activated and dependencies are installed.
Still stuck? Help me debug why uvicorn won't start my FastAPI app
Now send a sample error log to the endpoint. Before you do, think about this: what format do you expect Claude's response to be in? Structured JSON, or something else?
- Open a second terminal window (keep the server running in the first one).
- Send a sample database error log to your triage endpoint by running this command:
curl -s -X POST http://localhost:8000/triage \
-H "Content-Type: application/json" \
-d '{"log_text": "ERROR 2024-01-15 14:23:45 [db.pool] - Connection to PostgreSQL timed out after 30s. Active connections: 45/50. psycopg2.OperationalError: could not connect to server: Connection timed out", "source": "payment-service"}'
You should see a JSON response containing Claude's analysis of the error log. The response looks something like this:
{
"analysis": "**Severity: High**\n\nThis appears to be a database connection pool exhaustion issue...\n\n**Likely Causes:**\n- Connection pool near capacity (45/50)...\n\n**Recommended Actions:**\n1. Check database server health..."
}
Why does the response look like this?
Claude returns its analysis as free-form text. Sometimes it uses markdown formatting, sometimes bullet points, sometimes numbered lists. The exact format changes between requests.
This is fine for reading in a terminal. But if another service tried to parse this response programmatically, it would break. You'll fix this inconsistency in the next step.
🙋♀️ Getting an error instead of analysis?
- If you see a 422 error, check that your JSON body includes the log_text field spelled exactly as shown.
- If you see a 500 error, check the terminal running uvicorn for the full traceback. A common cause is a missing or invalid ANTHROPIC_API_KEY in your .env file.
- Make sure the uvicorn server is still running in your first terminal window.
Still stuck? Help me debug why my /triage endpoint is returning an error
You have a working API that accepts error logs and returns AI-powered analysis. The response is useful for humans reading it in a terminal. In the next step, you'll tame that free-text output into structured, predictable JSON that any downstream system can parse reliably.
Structure the AI Response
Your triage endpoint is live and returning analysis from Claude. You can send it any error log and get a response back in seconds.
There is a catch. Claude returns free-form text, and the format changes every time. One request gets bullet points, the next gets paragraphs, the next uses completely different headings. Any downstream system trying to parse these responses programmatically will break unpredictably.
In this step, get ready to:
- Observe how Claude's raw text output changes format across requests.
- Create Pydantic models and a structured system prompt that forces consistent JSON output.
- Update the endpoint to return validated, typed JSON every time.
Spot the format problem
Before building anything new, you need to see the problem with your own eyes. You will call the /triage endpoint twice with the exact same error log and compare the responses.
- Make sure your server is running. If it is not, start it by running:
uvicorn app.main:app --reload
- Open a second terminal window. Press Cmd+Space (macOS) or the Windows key (Windows), type Terminal, and press Enter.
- Send a test request to the triage endpoint by running this command:
curl -s -X POST http://localhost:8000/triage \
-H "Content-Type: application/json" \
-d '{"log_text": "ERROR 2024-01-15 10:23:45 - Database connection timeout after 30s. Host: db-primary.internal:5432. Connection pool exhausted (max=20, active=20, idle=0). Application: payment-service", "source": "payment-service"}'
- Copy the response and save it somewhere (a notes app or a scratch file) so you can compare it.
- Before you run the exact same command again, think about this: do you expect Claude to format both responses identically?
- Run the same curl command a second time and compare the two responses side by side.
The two responses are different. One might use headings like "Severity:" while the other uses "**Severity Level:**". One might return bullet points, the other numbered lists or full paragraphs.
This is the core problem. There is no reliable way to extract the severity level programmatically when the format changes on every request. This inconsistency is exactly what you are going to fix.
Why does Claude change format?
LLMs are probabilistic. Without explicit formatting instructions, Claude picks whatever text layout seems natural for the input. Small differences in how it interprets each request cascade into completely different output structures.
Define the response schema
The fix has two parts: Pydantic models that define the exact shape of every response, and a system prompt that tells Claude to return only JSON matching that shape. You will start with the models.
- Create a new file called models.py inside your app/ folder. Add the following code:
from enum import Enum
from pydantic import BaseModel, Field
# Constrain severity to exactly five levels
class Severity(str, Enum):
critical = "critical"
high = "high"
medium = "medium"
low = "low"
info = "info"
# Input model for the triage request
class TriageRequest(BaseModel):
log_text: str
source: str = "unknown"
# Structured output from Claude's analysis
class TriageAnalysis(BaseModel):
severity: Severity
category: str
summary: str
likely_causes: list[str]
recommended_actions: list[str]
# Full API response wrapping the analysis
class TriageResponse(BaseModel):
request_id: str
timestamp: str
analysis: TriageAnalysis
What does this code do?
- Severity is a string enum. It limits severity to exactly five values: critical, high, medium, low, and info. If Claude returns anything else, validation fails.
- TriageRequest replaces the inline model from main.py, keeping the same log_text and source fields.
- TriageAnalysis defines the exact JSON shape Claude must return. Every field has a specific type, so Pydantic will reject any response that does not match.
- TriageResponse wraps the analysis with a request_id and timestamp for tracing.
- Save app/models.py.
Now you need the code that actually calls Claude with a structured system prompt and validates the response against these models. This lives in a new triage.py module.
- Create a new file called triage.py inside your app/ folder. Add the imports and system prompt:
import logging
from anthropic import Anthropic
from app.config import Settings
from app.models import TriageAnalysis
logger = logging.getLogger("triage")
# Tell Claude exactly what JSON shape to return
SYSTEM_PROMPT = """You are an expert incident response engineer. Analyze the provided application log or error message and return a JSON object with exactly these fields:
{
"severity": "<critical|high|medium|low|info>",
"category": "<short category like 'memory', 'network', 'authentication', 'database', 'configuration', 'application'>",
"summary": "<one-sentence summary of the incident>",
"likely_causes": ["<cause 1>", "<cause 2>"],
"recommended_actions": ["<action 1>", "<action 2>"]
}
Rules:
- severity must be one of: critical, high, medium, low, info
- Return ONLY the JSON object, no markdown, no code fences, no explanation
- likely_causes and recommended_actions should each have 1-3 items"""
What does the system prompt do?
The system prompt is the key to consistent output. It gives Claude an exact JSON template to follow, names every allowed value for severity, and explicitly forbids markdown formatting or extra explanation. This is the "prompt engineering" half of the fix.
The other half is Pydantic validation. Even with a perfect prompt, an LLM can occasionally deviate. model_validate_json() catches any response that does not match the schema, so malformed output never reaches the caller.
- Confirm both new files exist by running this command in your second terminal:
ls app/
You should see __init__.py, config.py, main.py, models.py, and triage.py listed.
Missing a file?
Make sure you created both files inside the app/ folder, not the project root. The filenames should be models.py and triage.py (not model.py or triages.py).
help me find where my files ended up
- Now add the analyze_log function below the SYSTEM_PROMPT in app/triage.py:
def analyze_log(log_text: str, source: str, settings: Settings) -> TriageAnalysis:
# Create an Anthropic client with the API key and timeout
client = Anthropic(
api_key=settings.anthropic_api_key,
timeout=settings.claude_timeout,
)
# Send the log to Claude with the structured system prompt
message = client.messages.create(
model=settings.model_name,
max_tokens=settings.max_tokens,
system=SYSTEM_PROMPT,
messages=[
{
"role": "user",
"content": f"Source: {source}\n\nLog:\n{log_text}",
}
],
)
response_text = message.content[0].text
logger.info(
"Claude response received", extra={"response_length": len(response_text)}
)
# Validate Claude's JSON against our Pydantic model
analysis = TriageAnalysis.model_validate_json(response_text)
return analysis
What does this function do?
- analyze_log creates an Anthropic client and sends the error log to Claude along with the SYSTEM_PROMPT as a system parameter. This separates the formatting instructions from the actual log data.
- TriageAnalysis.model_validate_json() parses Claude's text response as JSON and validates every field against the TriageAnalysis model. If the response is missing a field or uses an invalid severity, Pydantic raises a ValidationError.
- Save app/triage.py.
✔️ Awesome, I've got everything!
Great. Make sure you have saved both app/models.py and app/triage.py.
ⓧ I'd like to double check the full code
app/triage.py is built from two chunks. Scroll up and confirm each one is present in order:
- Chunk 1: imports, logger, and the full SYSTEM_PROMPT string.
- Chunk 2: the analyze_log function that creates the client, calls messages.create(), and validates with model_validate_json().
Here is the full app/models.py for comparison:
from enum import Enum
from pydantic import BaseModel, Field
class Severity(str, Enum):
critical = "critical"
high = "high"
medium = "medium"
low = "low"
info = "info"
class TriageRequest(BaseModel):
log_text: str
source: str = "unknown"
class TriageAnalysis(BaseModel):
severity: Severity
category: str
summary: str
likely_causes: list[str]
recommended_actions: list[str]
class TriageResponse(BaseModel):
request_id: str
timestamp: str
analysis: TriageAnalysis
Wire up the structured endpoint
The models and triage logic are ready. Now you need to update app/main.py to use them instead of the raw Claude call.
- Replace the entire contents of app/main.py with the following code:
from datetime import datetime, timezone
from fastapi import FastAPI, Request
from app.config import Settings
from app.models import TriageRequest, TriageResponse
from app.triage import analyze_log
settings = Settings()
app = FastAPI(title="AI Incident Triage Assistant", version="1.0.0")
@app.post("/triage", response_model=TriageResponse)
async def triage_incident(payload: TriageRequest):
# Call the structured triage function
analysis = analyze_log(
log_text=payload.log_text,
source=payload.source,
settings=settings,
)
return TriageResponse(
request_id="placeholder",
timestamp=datetime.now(timezone.utc).isoformat(),
analysis=analysis,
)
What changed?
- The inline TriageRequest class is gone. It now lives in app/models.py alongside the other models.
- The direct Anthropic client call is replaced by analyze_log(), which handles the Claude call and Pydantic validation in one step.
- response_model=TriageResponse tells FastAPI to validate the outgoing response too, so the API docs show the exact shape callers can expect.
- request_id is set to "placeholder" for now. You will replace this with a real UUID in the next step.
- Save app/main.py.
✔️ Awesome, I've got everything!
Great. Double check you have saved app/main.py.
ⓧ I'd like to double check the full code
Here is the complete app/main.py:
from datetime import datetime, timezone
from fastapi import FastAPI, Request
from app.config import Settings
from app.models import TriageRequest, TriageResponse
from app.triage import analyze_log
settings = Settings()
app = FastAPI(title="AI Incident Triage Assistant", version="1.0.0")
@app.post("/triage", response_model=TriageResponse)
async def triage_incident(payload: TriageRequest):
analysis = analyze_log(
log_text=payload.log_text,
source=payload.source,
settings=settings,
)
return TriageResponse(
request_id="placeholder",
timestamp=datetime.now(timezone.utc).isoformat(),
analysis=analysis,
)
If you started the server with --reload, it should have restarted automatically when you saved. If it did not, stop it with Ctrl+C and start it again.
- Before you test, think: what do you expect the response to look like this time?
- Send the same test request again from your second terminal by running:
curl -s -X POST http://localhost:8000/triage \
-H "Content-Type: application/json" \
-d '{"log_text": "ERROR 2024-01-15 10:23:45 - Database connection timeout after 30s. Host: db-primary.internal:5432. Connection pool exhausted (max=20, active=20, idle=0). Application: payment-service", "source": "payment-service"}'
The response is now a structured JSON object with severity, category, summary, likely_causes, and recommended_actions every single time. Run it again and you will see the same fields in the same structure.
Seeing a 500 error or import error?
- Check your terminal running the server for the full traceback. An ImportError means one of the files is missing or has a typo in its filename.
- Make sure app/__init__.py still exists. Python needs it to treat app/ as a package.
- Verify that the old inline TriageRequest class and the direct Anthropic import are fully removed from main.py. Leftover imports will conflict with the new ones.
help me debug my structured triage endpoint
Your triage API now returns consistent, validated JSON on every request. No more guessing whether the severity is a heading, a bullet, or buried in a paragraph. Next up, you will add the production patterns that make this service debuggable and resilient: request tracing, structured logging, input validation, and timeout handling.
Add Production Hardening
Your triage endpoint works, but it is missing everything a production service needs to be debuggable, observable, and resilient. Right now there is no way to trace a request through your logs, no size limits on incoming payloads, and any Claude API failure crashes the endpoint with an unhandled 500 error.
In this step, you will add four hardening layers: request ID tracking, structured JSON logging, input validation with proper error codes, and fault-tolerant error handling for the Claude API.
In this step, get ready to:
- Add request ID middleware and structured JSON logging to every request.
- Add input validation with size constraints and a health check endpoint.
- Add error handling for Claude API timeouts and failures.
Add request ID middleware and structured JSON logging
When something goes wrong in a production API, the first question is always "which request caused this?" A unique ID on every request lets you trace a single call from arrival to response across all your log lines.
Structured JSON logs take this further. Instead of plain text lines that are hard to search, every log entry becomes a machine-parseable JSON object with consistent fields.
- In app/main.py, find the existing import block at the top of the file.
- Replace the entire import block (everything above settings = Settings()) with the following code:
import json
import logging
import sys
import uuid
from datetime import datetime, timezone
import anthropic
from fastapi import FastAPI, HTTPException, Request
from pydantic import ValidationError
from app.config import Settings
from app.models import TriageRequest, TriageResponse
from app.triage import analyze_log
What are these new imports for?
- json, logging, sys power the structured JSON logging system.
- uuid generates unique request IDs.
- anthropic, HTTPException, and ValidationError are for error handling, which you will wire up later in this step.
- Directly below the settings = Settings() line and above the existing app = FastAPI(...) line, add the JSONFormatter class and logger setup by pasting this code:
class JSONFormatter(logging.Formatter):
def format(self, record):
log_entry = {
"timestamp": datetime.now(timezone.utc).isoformat(),
"level": record.levelname,
"message": record.getMessage(),
"logger": record.name,
}
if hasattr(record, "request_id"):
log_entry["request_id"] = record.request_id
if record.exc_info and record.exc_info[0]:
log_entry["exception"] = self.formatException(record.exc_info)
for key in ("response_length", "source", "severity"):
if hasattr(record, key):
log_entry[key] = getattr(record, key)
return json.dumps(log_entry)
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(JSONFormatter())
logging.root.handlers = [handler]
logging.root.setLevel(settings.log_level)
logger = logging.getLogger("app")
What does this code do?
The JSONFormatter class overrides Python's default log formatter. Instead of plain text lines, every log entry becomes a JSON object with a timestamp, level, message, and logger name.
It also checks for extra fields like request_id, response_length, and severity. If any of these were passed via the extra keyword in a logger.info() call, they get included in the JSON output automatically.
The four lines after the class create a StreamHandler that writes to stdout, attach the JSON formatter to it, and replace the root logger's handlers so ALL loggers in the app output structured JSON.
- Save app/main.py.
- Restart the server by running:
uvicorn app.main:app --reload
- In a second terminal, send a triage request by running:
curl -s -X POST http://localhost:8000/triage \
-H "Content-Type: application/json" \
-d '{"log_text": "ERROR: Connection to database timed out after 30s at db.connect()"}'
Check the terminal running uvicorn. You should see a structured JSON log line from the triage logger containing "message": "Claude response received" and a response_length field, instead of a plain text log line.
Still seeing plain text logs?
Make sure the four handler-setup lines (handler = logging.StreamHandler(sys.stdout) through logging.root.setLevel(settings.log_level)) are at the module level, not inside a function. They need to run when the module loads.
ask me to help debug my structured logging setup
- Now replace the existing app = FastAPI(title="AI Incident Triage Assistant", version="1.0.0") line with the following code, which reformats the app definition and adds the request ID middleware below it:
app = FastAPI(
title="AI Incident Triage Assistant",
version="1.0.0",
)
@app.middleware("http")
async def add_request_id(request: Request, call_next):
request_id = str(uuid.uuid4())
request.state.request_id = request_id
logger.info(
f"Incoming {request.method} {request.url.path}",
extra={"request_id": request_id},
)
response = await call_next(request)
response.headers["X-Request-ID"] = request_id
return response
What does the middleware do?
This middleware intercepts every HTTP request before it reaches your endpoint. It generates a UUID, stores it in request.state.request_id, logs the incoming method and path, then passes the request on. After the endpoint responds, it attaches the UUID as an X-Request-ID response header.
Any caller can read this header to correlate their request with your server logs.
The endpoint also needs to use the request ID the middleware generates instead of the "placeholder" string.
- In app/main.py, find the /triage endpoint function. Find this line:
async def triage_incident(payload: TriageRequest):
- Replace it with this line to accept the Request object from the middleware:
async def triage_incident(request: Request, payload: TriageRequest):
- Next, find this line inside the return statement:
request_id="placeholder",
- Replace "placeholder" with request_id so it uses the real UUID from the middleware. Also add request_id = request.state.request_id as the first line inside the function body, so the line looks like this:
request_id=request_id,
- Save app/main.py. The server should auto-reload.
- Send another triage request and check for the X-Request-ID header by running:
curl -s -i -X POST http://localhost:8000/triage \
-H "Content-Type: application/json" \
-d '{"log_text": "ERROR: Connection to database timed out after 30s at db.connect()"}'
You should see an x-request-id header in the response headers containing a UUID. In the uvicorn terminal, the JSON log lines now include a matching request_id field.
Why use the -i flag?
The -i flag tells curl to include the response headers in the output. Without it, you would only see the JSON body and miss the X-Request-ID header entirely.
🙋♀️ Don't see the X-Request-ID header?
Make sure you added request: Request as the first parameter of triage_incident (before payload). The middleware stores the request ID on request.state, so the endpoint needs the Request object to access it.
help me debug my request ID middleware
Add input validation and a health endpoint
Right now, a caller could send an empty string or a 10MB log dump and your endpoint would happily forward it to Claude. Input validation prevents both extremes.
- Open app/models.py and find the TriageRequest class.
- Replace the entire TriageRequest class with this version that adds size constraints:
class TriageRequest(BaseModel):
log_text: str = Field(
...,
min_length=10,
max_length=5000,
description="The error log or incident text to analyze",
)
source: str = Field(
default="unknown", description="Source system or application name"
)
What does this code do?
Pydantic's Field function lets you set constraints directly on model fields. min_length=10 rejects logs that are too short to be meaningful. max_length=5000 prevents oversized payloads from running up your Claude API costs.
FastAPI automatically returns HTTP 422 with a clear validation error message when a request violates these constraints. You don't need to write any error handling code for this.
- Save app/models.py.
- Test the input validation by sending a log that is too short (fewer than 10 characters):
curl -s -X POST http://localhost:8000/triage \
-H "Content-Type: application/json" \
-d '{"log_text": "error"}'
You should see a 422 response with a detail array explaining that log_text must be at least 10 characters long.
Your service also needs a health endpoint that container orchestrators can poll to check if the service is alive.
- Switch back to app/main.py. Add the /health endpoint directly above the existing @app.post("/triage", ...) decorator by pasting this code:
@app.get("/health")
async def health_check():
return {"status": "healthy", "service": "incident-triage-assistant"}
What is a health endpoint?
A /health endpoint returns a simple JSON response confirming the service is running. Container orchestrators like Kubernetes and Docker Compose use these to decide whether to route traffic to a container or restart it.
- Save app/main.py.
- Test the health endpoint by running:
curl -s http://localhost:8000/health
You should see {"status":"healthy","service":"incident-triage-assistant"} in the response.
Getting a 404 on /health?
Make sure the @app.get("/health") decorator is placed after the middleware function and before the @app.post("/triage", ...) decorator. If it is placed before the app = FastAPI(...) line, the app variable will not exist yet and you will get a NameError.
help me fix my /health endpoint
Add error handling for Claude API failures
If the Claude API times out or returns an error, your endpoint currently crashes with an unhandled 500. Production services need to catch these failures and return clear, actionable HTTP status codes instead.
- In app/main.py, find the entire triage_incident function (from @app.post("/triage", ...) to the end of the file).
- Replace the entire function with this version that adds request logging and wraps the Claude call in error handling:
@app.post("/triage", response_model=TriageResponse)
async def triage_incident(request: Request, payload: TriageRequest):
request_id = request.state.request_id
logger.info(
"Processing triage request",
extra={"request_id": request_id, "source": payload.source},
)
try:
analysis = analyze_log(
log_text=payload.log_text,
source=payload.source,
settings=settings,
)
except anthropic.APITimeoutError:
logger.error("Claude API timed out", extra={"request_id": request_id})
raise HTTPException(status_code=504, detail="AI analysis timed out. Please try again.")
except anthropic.APIStatusError as e:
logger.error(f"Claude API error: {e.status_code}", extra={"request_id": request_id}, exc_info=True)
raise HTTPException(status_code=502, detail="AI service returned an error.")
except ValidationError as e:
logger.error(f"Failed to parse Claude response: {e}", extra={"request_id": request_id}, exc_info=True)
raise HTTPException(status_code=502, detail="AI returned an unparseable response.")
except Exception as e:
logger.error(f"Unexpected triage failure: {e}", extra={"request_id": request_id}, exc_info=True)
raise HTTPException(status_code=502, detail="AI analysis failed. Please try again.")
What does each except block handle?
- anthropic.APITimeoutError fires when the Claude API takes too long. Returns HTTP 504 (gateway timeout).
- anthropic.APIStatusError catches non-200 responses from Claude's API (rate limits, server errors). Returns HTTP 502 (bad gateway).
- ValidationError catches cases where Claude returns text that does not match your TriageAnalysis schema. Returns HTTP 502.
- The fallback Exception catches anything unexpected so the endpoint never returns an unhandled 500.
Every error is logged with the request_id so you can match a caller's error response to the exact log entry that explains what went wrong.
- Directly below the last except block, add the success-path logging and return statement to complete the function:
logger.info(
"Triage complete",
extra={"request_id": request_id, "severity": analysis.severity.value},
)
return TriageResponse(
request_id=request_id,
timestamp=datetime.now(timezone.utc).isoformat(),
analysis=analysis,
)
What does this code do?
After a successful analysis, the function logs the severity that Claude assigned (so you can track severity distribution across requests). Then it returns the full TriageResponse with the real request_id from the middleware and a UTC timestamp.
- Save app/main.py.
Before you test the error handling, take a moment to predict what will happen. If you set the Claude API timeout to an impossibly short value like 0.01 seconds, what HTTP status code do you think the endpoint will return?
- Open app/config.py and find the claude_timeout line. Temporarily change the value to test timeout handling:
claude_timeout: float = 0.01
- Save app/config.py and let the server reload.
- Send a triage request to trigger the timeout:
curl -s -X POST http://localhost:8000/triage \
-H "Content-Type: application/json" \
-d '{"log_text": "ERROR: Connection to database timed out after 30s at db.connect()"}'
You should see a 504 response with {"detail":"AI analysis timed out. Please try again."}. The Anthropic SDK retries timeouts twice by default, so this request may take a few seconds before returning.
Why does the timeout take a few seconds?
The Anthropic SDK automatically retries timed-out requests 2 times with a short exponential backoff. Even with a 0.01-second timeout, the SDK attempts the call 3 times total before raising APITimeoutError.
- Now change the timeout back to its production value in app/config.py:
claude_timeout: float = 30.0
- Save app/config.py.
Getting a different error instead of 504?
If you see a 500 error or a traceback, make sure you imported anthropic (the module, not just Anthropic the class) at the top of app/main.py. The except block catches anthropic.APITimeoutError, which requires the module-level import.
help me debug my Claude API error handling
Time to verify all four hardening layers together. What do you expect to see when you send a valid triage request now?
- Send a final triage request with verbose headers by running:
curl -s -i -X POST http://localhost:8000/triage \
-H "Content-Type: application/json" \
-d '{"log_text": "FATAL: java.lang.OutOfMemoryError: Java heap space at com.app.Worker.process(Worker.java:142)"}'
You should see the x-request-id header in the response. In the uvicorn terminal, you should see structured JSON log lines with matching request_id fields.
✔️ Awesome, I've got everything!
Great. Double check you have saved both app/main.py and app/models.py.
ⓧ I'd like to double check the full code
Your app/models.py should now look like this:
from enum import Enum
from pydantic import BaseModel, Field
class Severity(str, Enum):
critical = "critical"
high = "high"
medium = "medium"
low = "low"
info = "info"
class TriageRequest(BaseModel):
log_text: str = Field(
...,
min_length=10,
max_length=5000,
description="The error log or incident text to analyze",
)
source: str = Field(
default="unknown", description="Source system or application name"
)
The TriageAnalysis and TriageResponse classes below should be unchanged from the previous step.
For app/main.py, scroll through your file and confirm these sections appear in order:
- The updated import block with json, logging, sys, uuid, anthropic, HTTPException, and ValidationError.
- settings = Settings() followed by the JSONFormatter class.
- The logger handler setup (StreamHandler to stdout).
- The app = FastAPI(...) definition followed by the add_request_id middleware.
- The health_check endpoint, then the full triage_incident endpoint with the try/except block.
Your triage API now has real production bones: every request is traceable, every log is structured, bad input is rejected before it costs you an API call, and Claude failures return clear status codes instead of unhandled crashes. Next up, you will containerize the whole thing with Docker so it runs anywhere.
Containerize with Docker
Your triage service now has production-grade patterns in place. Request IDs, structured logging, input validation, timeout handling, and a health endpoint.
The service still depends on your local Python environment and .env file. If you handed this project to a colleague, they would need to install the same Python version, create a virtual environment, install dependencies, and configure environment variables before anything works. Containerizing the service with Docker packages everything into a single, portable image that runs anywhere.
In this step, get ready to:
- Create a Dockerfile and .dockerignore for the project.
- Build and run the service inside a Docker container.
- Verify both endpoints work through the container.
Create the Dockerfile
A Dockerfile is a recipe that tells Docker what base image to start from, what files to copy in, what to install, and how to start the app. Docker reads this file and builds a self-contained image from it.
- Create a new file called Dockerfile (no file extension) in the root of your project folder, at the same level as requirements.txt.
- Paste the following code into it:
FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY app/ app/
EXPOSE 8000
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]
What does this Dockerfile do?
- FROM python:3.12-slim starts from a lightweight Python 3.12 image with minimal OS packages.
- WORKDIR /app sets the working directory inside the container. All following commands run from here.
- COPY requirements.txt . followed by RUN pip install installs dependencies first. This layer gets cached by Docker, so rebuilds are fast when only your app code changes.
- COPY app/ app/ copies your application code into the container.
- CMD tells Docker how to start uvicorn when the container runs, binding to 0.0.0.0 so it accepts connections from outside the container.
- Save Dockerfile.
✔️ Awesome, I've got everything!
Great. Make sure you saved your file before moving on.
ⓧ I'd like to double check the full code
FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY app/ app/
EXPOSE 8000
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]
Next, you need a .dockerignore file. This tells Docker which files to exclude when building the image. You do not want your .env file (which holds your API key) baked into the image.
- Create a new file called .dockerignore in the root of your project folder (same level as Dockerfile).
- Paste the following content:
__pycache__
*.pyc
.env
.git
.venv
Why exclude these files?
Your .env file contains your API key. Baking secrets into an image means anyone with access to the image can extract them. Instead, you will pass the key at runtime using the -e flag.
The other entries (__pycache__, .git, .venv) are local artifacts that would bloat the image without adding value.
- Save .dockerignore.
- Confirm both files exist by running:
ls
You should see Dockerfile and .dockerignore listed alongside your existing files like requirements.txt and the app/ directory.
Don't see Dockerfile or .dockerignore?
- Make sure you created the files in the project root (the same folder that contains requirements.txt), not inside the app/ directory.
- The file must be named exactly Dockerfile with a capital D and no extension. Not dockerfile.txt or Dockerfile.txt.
- Files starting with a dot (like .dockerignore) are hidden by default on macOS and Linux. Run ls -a to see hidden files.
Why aren't my Dockerfile and .dockerignore showing up?
Build and run the container
With the Dockerfile and .dockerignore in place, you are ready to build the image and run it. The container needs your API key at runtime, so you will pass it as an environment variable instead of baking it into the image.
- If your local uvicorn server is still running from the previous step, stop it by pressing Ctrl+C in that terminal.
- Make sure Docker Desktop is running. You can check by looking for the Docker whale icon in your menu bar (macOS) or system tray (Windows).
Your .env file will not be inside the container (you excluded it in .dockerignore). You need to export your API key to the terminal so the docker run command can pass it in.
- Open your .env file and copy the value of ANTHROPIC_API_KEY.
- Export it in your terminal by running the command below, replacing your-api-key-here with your actual key:
export ANTHROPIC_API_KEY=your-api-key-here
Windows PowerShell users
If you are using PowerShell instead of Git Bash, use this syntax instead:
$env:ANTHROPIC_API_KEY = "your-api-key-here"
And in the docker run command later, replace $ANTHROPIC_API_KEY with $env:ANTHROPIC_API_KEY.
- Build the Docker image by running this command:
docker build -t triage-assistant .
Docker will pull the python:3.12-slim base image, install your dependencies, and copy your app code. This may take a minute or two the first time. You should see FINISHED in the output when it completes.
Build failed?
- Make sure Docker Desktop is running. The docker command will not work if the Docker daemon is not started.
- Make sure you are running the command from your project root (the folder containing Dockerfile). The . at the end tells Docker to use the current directory as its build context.
- If you see a pip install error, double-check that your requirements.txt has the correct package names and versions.
Why is my Docker build failing?
- Start the container by running this command:
docker run -p 8000:8000 -e ANTHROPIC_API_KEY=$ANTHROPIC_API_KEY triage-assistant
What do these flags do?
- -p 8000:8000 maps port 8000 on your machine to port 8000 inside the container. This lets you reach the service at localhost:8000 just like before.
- -e ANTHROPIC_API_KEY=$ANTHROPIC_API_KEY passes your exported API key into the container as an environment variable. The Settings class in config.py reads it automatically.
This command runs in the foreground, so the terminal will show the uvicorn startup logs and any request logs. You need a second terminal to send requests.
- Open a new terminal tab by pressing Cmd+T (macOS) or Ctrl+Shift+T (Windows).
- Test the health endpoint by running:
curl http://localhost:8000/health
You should see:
{"status":"healthy","service":"incident-triage-assistant"}
The health endpoint is responding from inside the container. Now for the real test.
Before you run the next command, think about this: the container has no .env file inside it. Do you think the /triage endpoint will successfully reach Claude and return a structured analysis, or will it fail?
- Send a triage request to the containerized service by running:
curl -X POST http://localhost:8000/triage \
-H "Content-Type: application/json" \
-d '{"log_text": "ERROR 2024-01-15 10:23:45 Connection refused to database host db-primary:5432 after 3 retries. java.net.ConnectException: Connection refused", "source": "payment-service"}'
You should see a full structured JSON response with severity, category, summary, likely_causes, and recommended_actions. The -e flag passed your API key into the container, so pydantic-settings picks it up from the environment even without a .env file.
Check your container logs
Switch back to the terminal tab where the container is running. You should see structured JSON log lines from your JSONFormatter. Each log entry includes a request_id field, just like when you ran the server locally.
🙋♀️ Not getting a response from the container?
- If you see a connection refused error, make sure you stopped your local uvicorn server before running the container. Both cannot use port 8000 at the same time.
- If you see a 502 error mentioning the AI service, check that you exported ANTHROPIC_API_KEY in the same terminal session where you ran docker run. Verify the key is set by running echo $ANTHROPIC_API_KEY in a new terminal tab.
- If the container exits immediately, check the logs in that terminal for a startup error. A missing or empty ANTHROPIC_API_KEY will cause a Pydantic validation error on startup.
Help me debug my Docker container for the triage API.
Secret mission
Add Deterministic Escalation Rules
Your triage API asks Claude to rate severity, but AI assessments are probabilistic. The same OutOfMemoryError might get rated "high" one time and "critical" the next. In this mission, you'll add deterministic escalation rules that guarantee certain patterns always trigger a specific severity, regardless of what the AI thinks.
Clean Up Your Resources
Clean Up Your Resources
This project runs locally and uses the Anthropic API on a pay-per-use basis. There are no ongoing costs once you stop making API calls, but you may want to clean up local Docker resources and project files.
Resources you used:
- Docker container created from the triage-assistant image.
- Docker image triage-assistant.
- Project folder on your Desktop containing all source files, the .venv virtual environment, and the .env file with your Anthropic API key.
Decide whether to keep your resources running, pause them to come back later, or delete them entirely.
Keep everything running
No action needed. Choose this if you are still actively building on the triage assistant or plan to extend it soon.
- Your Docker container may still be running. It will continue to use system memory while active.
- The Anthropic API charges only per request. There are no ongoing costs while the service is idle.
- Docker Desktop will continue using system resources while it is open.
- Your Anthropic API key is stored in the .env file inside your project folder. Keep this file safe and do not commit it to version control.
Pause - I'll come back to this later
Shut down running processes to free up memory. Your files and Docker image stay on disk so you can resume quickly.
- Stop the running Docker container by pressing Ctrl+C in the terminal where it is running. If the terminal is closed, find and stop the container by running:
docker ps
docker stop <container_id>
- Close Docker Desktop to free up system memory.
- Your project folder, Docker image, and .env file remain on disk. To resume, open Docker Desktop, then run the container again.
Delete - I don't want to use this again
Remove all project resources from your machine and start fresh.
- Stop and remove the Docker container by running:
docker ps -a
docker rm -f <container_id>
- Remove the triage-assistant Docker image by running:
docker rmi triage-assistant
- Delete the project folder from your Desktop. On macOS or Linux, run:
cd ~/Desktop
rm -rf incident-triage-assistant
On Windows, run:
cd ~\Desktop
rmdir /s /q incident-triage-assistant
What about my Anthropic API key?
Deleting the project folder removes the .env file that held your key locally. If you no longer need the key, you can also revoke it in the Anthropic console.
Open your API keys page in the Claude Console, click the three-dot menu next to your key, and select Delete API Key.
Nice Work!
Nice Work!
You just built a production-style incident triage API that turns messy error logs into clean, actionable analysis. Paste in a stack trace, get back structured JSON telling you exactly what went wrong and what to do about it.
You've learned how to:
- Build a FastAPI service with Pydantic-validated request and response models that accept error logs via POST /triage and return structured AI-powered triage analysis using Claude Haiku 4.5 via the Anthropic Python SDK.
- Enforce consistent JSON output from an LLM using a structured system prompt paired with Pydantic validation, guaranteeing typed fields like severity, category, and recommended actions on every response.
- Add production hardening patterns including request ID middleware with X-Request-ID headers, structured JSON logging, input size validation, Claude API timeout handling with granular HTTP status codes, a /health endpoint, and containerize the entire service with Docker.
- Secret Mission: Complete an optional challenge to push your skills further.
Ready to quiz yourself?