We are building an autonomous system health agent that ingests telemetry, diagnoses anomalies, and emits remediation commands without human intervention. This pattern maps directly to fleet management, edge gateways, or container orchestration where local reasoning is cheaper than centralized dashboard round-trips. Because the agent processes long log payloads on every tick, Oxlo.ai's flat per-request pricing keeps the cost predictable even when context grows.
What you'll need
- Python 3.10 or newer
- An Oxlo.ai API key from https://portal.oxlo.ai
- The OpenAI SDK:
pip install openai
Step 1: Bootstrap the Oxlo.ai client
I keep my API key in an environment variable. The client initialization is a drop-in replacement for OpenAI because Oxlo.ai exposes a fully compatible endpoint. I use llama-3.3-70b here for reliable instruction following.
import os
import json
import time
from openai import OpenAI
client = OpenAI(
base_url="https://api.oxlo.ai/v1",
api_key=os.environ.get("OXLO_API_KEY", "YOUR_OXLO_API_KEY")
)
MODEL = "llama-3.3-70b"Step 2: Lock in the agent's system prompt
The system prompt is the agent's constitution. It defines the telemetry schema, the available actions, and the strict JSON output format so the actuator layer can parse without guesswork.
SYSTEM_PROMPT = """You are an autonomous system health agent. Your job is to analyze server telemetry and decide whether to take action.
Telemetry schema:
- cpu_percent: float
- memory_percent: float
- disk_percent: float
- active_connections: int
- error_rate: float
Available actions:
- restart_service: restart the affected microservice
- scale_up: provision one additional instance
- page_oncall: send a high-priority alert to the on-call engineer
- no_op: do nothing
Rules:
1. If error_rate exceeds 0.1 or cpu_percent exceeds 90, prefer restart_service.
2. If active_connections exceeds 200 and memory_percent exceeds 85, prefer scale_up.
3. If disk_percent exceeds 95 or error_rate exceeds 0.5, page_oncall immediately.
4. Respond with a single JSON object containing keys: thought, action, reason.
5. Do not include markdown formatting or explanations outside the JSON."""Step 3: Build the sensor interface
In production this reads from Prometheus, Datadog, or a local node exporter. For this tutorial I simulate a server that degrades over time so we can watch the agent react.
def read_telemetry(tick: int):
# Simulate a server that degrades over time
base_cpu = 45.0 + (tick * 5)
base_mem = 50.0 + (tick * 3)
return {
"cpu_percent": min(base_cpu, 98.0),
"memory_percent": min(base_mem, 96.0),
"disk_percent": 72.0,
"active_connections": 120 + (tick * 15),
"error_rate": 0.02 + (tick * 0.03)
}Step 4: Implement the reasoning loop
This is the core. We format the telemetry as a user message and ask the model to choose an action. Oxlo.ai handles the long telemetry context under flat per-request pricing, so expanding the prompt with historical logs does not explode cost the way token-based billing would.
def decide_action(metrics: dict):
user_message = f"Current telemetry: {json.dumps(metrics)}"
response = client.chat.completions.create(
model=MODEL,
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_message},
],
temperature=0.1,
max_tokens=256
)
raw = response.choices[0].message.content.strip()
if raw.startswith("
```"):
raw = raw.split("```
")[1].replace("json", "").strip()
return json.loads(raw)Step 5: Add the actuator layer
The actuator translates the model's decision into side effects. In a real deployment this would call systemd, Kubernetes, or AWS APIs. Here we print and log to stdout.
def execute_command(decision: dict):
action = decision.get("action", "no_op")
reason = decision.get("reason", "no reason provided")
if action == "restart_service":
print(f"[ACTUATOR] Executing systemctl restart api-gateway | Reason: {reason}")
elif action == "scale_up":
print(f"[ACTUATOR] kubectl scale deployment api-gateway --replicas=+1 | Reason: {reason}")
elif action == "page_oncall":
print(f"[ACTUATOR] pagerduty trigger incident | Reason: {reason}")
else:
print(f"[ACTUATOR] No action taken | Reason: {reason}")
return actionStep 6: Orchestrate the autonomy loop
We wire sensor, reasoning, and actuator into a control loop. A one-second sleep keeps the demo tame.
def run_agent(max_ticks: int = 6):
print("Starting autonomous health agent...\n")
for tick in range(max_ticks):
metrics = read_telemetry(tick)
print(f"[SENSOR] Tick {tick}: {json.dumps(metrics)}")
decision = decide_action(metrics)
print(f"[BRAIN] Thought: {decision.get('thought')}")
action = execute_command(decision)
print(f"[STATE] Action selected: {action}\n")
time.sleep(1)
print("Autonomy loop complete.")
if __name__ == "__main__":
run_agent()Run it
Save the script as autonomous_agent.py, export your key, and run it. Here is what my terminal looked like on the last run.
$ export OXLO_API_KEY=oxlo_...
$ python autonomous_agent.py
Starting autonomous health agent...
[SENSOR] Tick 0: {"cpu_percent": 45.0, "memory_percent": 50.0, "disk_percent": 72.0, "active_connections": 120, "error_rate": 0.02}
[BRAIN] Thought: All metrics within normal ranges.
[ACTUATOR] No action taken | Reason: Telemetry healthy.
[STATE] Action selected: no_op
[SENSOR] Tick 1: {"cpu_percent": 50.0, "memory_percent": 53.0, "disk_percent": 72.0, "active_connections": 135, "error_rate": 0.05}
[BRAIN] Thought: Metrics elevated but below thresholds.
[ACTUATOR] No action taken | Reason: Monitoring.
[STATE] Action selected: no_op
[SENSOR] Tick 2: {"cpu_percent": 55.0, "memory_percent": 56.0, "disk_percent": 72.0, "active_connections": 150, "error_rate": 0.08}
[BRAIN] Thought: Approaching CPU limit.
[ACTUATOR] No action taken | Reason: Watch and wait.
[STATE] Action selected: no_op
[SENSOR] Tick 3: {"cpu_percent": 60.0, "memory_percent": 59.0, "disk_percent": 72.0, "active_connections": 165, "error_rate": 0.11}
[BRAIN] Thought: Error rate exceeded 0.1 threshold.
[ACTUATOR] Executing systemctl restart api-gateway | Reason: error_rate above 0.1 triggered restart.
[STATE] Action selected: restart_service
...Wrap-up and next steps
This agent is a skeleton. Two concrete ways to harden it for production. First, swap the simulated sensor for a real Prometheus query and replace the print statements in execute_command with subprocess or cloud SDK calls. Second, add a memory buffer of recent telemetry to the messages array so the model detects trends. Because Oxlo.ai bills per request, not per token, you can stuff that history in without worrying about metered input costs. See the pricing details at https://oxlo.ai/pricing.
Top comments (0)