When production systems throw errors, you need structured incident data without shipping every log line to a third-party observability platform. In this tutorial, I will build a FastAPI service that runs on your own server and uses Oxlo.ai to turn raw application logs into classified, triaged JSON reports. Because Oxlo.ai uses flat per-request pricing, long log dumps do not inflate your bill the way token-based providers do, which makes deep-context log analysis practical.
What you'll need
- Python 3.10 or newer
- An Oxlo.ai API key from https://portal.oxlo.ai
pip install openai fastapi uvicorn pydantic python-dotenv- A sample log file, or you can paste the string I use below
Step 1: Configure the Oxlo.ai client
I load the API key from the environment and initialize the OpenAI-compatible client pointing at Oxlo.ai. I use Llama 3.3 70B as the general-purpose flagship for reliable structured output.
import os
from openai import OpenAI
from dotenv import load_dotenv
load_dotenv()
client = OpenAI(
base_url="https://api.oxlo.ai/v1",
api_key=os.getenv("OXLO_API_KEY")
)
MODEL = "llama-3.3-70b"Step 2: Define the schema and system prompt
The agent must return predictable JSON, so I define a Pydantic model and a strict system prompt that forces the model to classify severity and extract root cause.
from pydantic import BaseModel, Field
from typing import List
class IncidentReport(BaseModel):
severity: str = Field(description="One of: CRITICAL, HIGH, MEDIUM, LOW")
category: str = Field(description="e.g., Database, Network, Memory, Auth")
root_cause: str = Field(description="One-sentence technical root cause")
affected_services: List[str] = Field(description="List of affected service names")
recommended_action: str = Field(description="Immediate fix or investigation step")The system prompt is the most important part. I keep it separate so you can tune it without touching the route code.
SYSTEM_PROMPT = """You are a senior site-reliability engineer analyzing application logs.
Read the provided logs and output a single JSON object matching the required schema.
Rules:
- severity must be one of CRITICAL, HIGH, MEDIUM, LOW.
- category must be the single best-fitting technical domain.
- root_cause must be one sentence and specific.
- affected_services must be an array of strings.
- recommended_action must be a concrete, immediate step.
Do not include markdown code fences, only raw JSON."""Step 3: Build the analysis function
This function takes raw log text, sends it to Oxlo.ai, and parses the JSON response. I strip accidental markdown fences because some models wrap JSON in triple backticks despite instructions.
import json
def analyze_logs(log_text: str) -> dict:
response = client.chat.completions.create(
model=MODEL,
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": f"Analyze these logs and return raw JSON only:\n\n{log_text}"},
],
)
content = response.choices[0].message.content.strip()
if content.startswith("
```"):
content = content.split("\n", 1)[1].rsplit("```
", 1)[0].strip()
return json.loads(content)Step 4: Wrap it in a FastAPI server
Now I expose the analyzer as a POST endpoint so other services on your network can send logs to it. FastAPI handles validation and async concurrency for us.
from fastapi import FastAPI, HTTPException
app = FastAPI(title="Log Triage Agent")
class LogRequest(BaseModel):
service_name: str
log_payload: str
@app.post("/analyze")
async def analyze_endpoint(req: LogRequest):
try:
result = analyze_logs(req.log_payload)
return {
"service": req.service_name,
"incident": result
}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))Step 5: Add batch processing
For production use, you often want to analyze multiple log streams at once. I add a /batch endpoint that processes them concurrently using asyncio.gather so a single slow response does not block the rest.
import asyncio
from typing import List
class BatchLogRequest(BaseModel):
items: List[LogRequest]
async def analyze_async(log_text: str) -> dict:
loop = asyncio.get_event_loop()
return await loop.run_in_executor(None, analyze_logs, log_text)
@app.post("/batch")
async def batch_endpoint(req: BatchLogRequest):
tasks = [analyze_async(item.log_payload) for item in req.items]
results = await asyncio.gather(*tasks, return_exceptions=True)
output = []
for item, res in zip(req.items, results):
if isinstance(res, Exception):
output.append({"service": item.service_name, "error": str(res)})
else:
output.append({"service": item.service_name, "incident": res})
return {"results": output}Run it
Start the server with Uvicorn.
uvicorn main:app --host 0.0.0.0 --port 8000Test the /analyze endpoint with cURL using a realistic log snippet.
curl -X POST http://localhost:8000/analyze \
-H "Content-Type: application/json" \
-d '{
"service_name": "payment-gateway",
"log_payload": "2024-05-21T14:32:10Z ERROR connection timeout to db-primary.internal:5432 after 30s\n2024-05-21T14:32:11Z WARN fallback to db-replica triggered\n2024-05-21T14:32:12Z ERROR payment webhook failed: could not acquire lock"
}'You should receive structured output similar to this.
{
"service": "payment-gateway",
"incident": {
"severity": "CRITICAL",
"category": "Database",
"root_cause": "Primary database connection timeout caused lock acquisition failure in the payment webhook handler.",
"affected_services": ["payment-gateway", "db-primary"],
"recommended_action": "Check db-primary network connectivity and connection pool exhaustion."
}
}Wrap-up and next steps
This agent is already useful as a lightweight internal API, but two upgrades will make it production ready. First, put the Oxlo.ai calls behind a queue like Celery or Redis Streams so heavy log bursts do not exhaust your FastAPI worker pool. Second, if you are processing multilingual logs or need deeper reasoning for distributed traces, swap the model to qwen-3-32b or kimi-k2.6. Both are available on Oxlo.ai with the same flat per-request pricing, so longer contexts and reasoning chains do not increase your cost. For details on request limits and enterprise options, see https://oxlo.ai/pricing.
Top comments (0)