"""Diagnose a server with a fixed list of read-only checks."""
import os
import subprocess
import sys
import openai
client = openai.OpenAI(api_key=os.environ["ORCA_API_KEY"], base_url=os.environ["ORCA_BASE_URL"])
agents = client.beta.agents
sessions = agents.sessions
HOST = sys.argv[1] if len(sys.argv) > 1 else "web-1"
QUESTION = sys.argv[2] if len(sys.argv) > 2 else "The site is slow. What is going on?"
# The only commands that can ever run. The model picks a name, never a command.
CHECKS = {
"uptime": "uptime",
"memory": "free -m",
"disk": "df -h -x tmpfs -x devtmpfs",
"top_cpu": "ps -eo pid,user,%cpu,%mem,etime,comm --sort=-%cpu | head -15",
"top_memory": "ps -eo pid,user,%cpu,%mem,etime,comm --sort=-%mem | head -15",
"listening_ports": "ss -ltnp",
"failed_services": "systemctl --failed --no-pager",
"recent_errors": "journalctl -p err --since '-1h' --no-pager | tail -50",
}
INSTRUCTIONS = """\
You diagnose Linux servers. You cannot run commands yourself.
Call run_check with one of: uptime, memory, disk, top_cpu, top_memory, listening_ports,
failed_services, recent_errors. Start broad (uptime, memory, disk), then narrow down.
Base every conclusion on check output; quote the line that supports it.
If the checks cannot answer the question, say which information is missing.
Suggest fixes as commands for a person to run, and say which ones are risky,
such as restarting a service or killing a process. Do not use emojis."""
RUN_CHECK = {
"type": "function",
"name": "run_check",
"description": "Run one named read-only check on the server and return its output.",
"parameters": {
"type": "object",
"properties": {"check": {"type": "string", "enum": sorted(CHECKS)}},
"required": ["check"],
"additionalProperties": False,
},
"defer_loading": False,
}
def run_check(args):
command = CHECKS.get(str(args.get("check")))
if command is None:
return {"error": f"Unknown check. Use one of: {', '.join(sorted(CHECKS))}"}
try:
result = subprocess.run(
["ssh", "-o", "BatchMode=yes", HOST, command],
capture_output=True, text=True, errors="replace", timeout=30,
)
except subprocess.TimeoutExpired:
return {"error": "The check timed out after 30 seconds."}
except OSError as error:
return {"error": f"Could not run ssh: {error}"}
# Cap the output: the model pays for every token it reads.
return {"exit_code": result.returncode, "output": (result.stdout + result.stderr)[-8000:]}
agent = session = None
try:
agent = agents.create(
model=os.environ["ORCA_MODEL"], name="server-health", instructions=INSTRUCTIONS, tools=[RUN_CHECK],
)
session = sessions.create(agent_id=agent.id, environment={"type": "none"})
status = None
with sessions.stream(session.id, input=f"Host {HOST}: {QUESTION}",
tool_handlers={"run_check": run_check}) as stream:
for event in stream: # read to the end, or the helper stops answering calls
if event.type == "agent.session.turn.completed":
status = event.turn.status
if status != "completed":
raise SystemExit(f"Diagnosis did not complete: {status}")
for item in sessions.items.list(session.id, limit=50).data:
if item.type == "message" and item.role == "assistant":
print(" ".join(p.text for p in item.content or [] if getattr(p, "text", None)))
break
finally:
# Delete each resource on its own, so one failed delete does not skip the rest.
if session:
try:
sessions.delete(session.id)
except openai.APIError:
pass
if agent:
try:
agents.delete(agent.id)
except openai.APIError:
pass
client.close()
