#!/usr/bin/env python3
"""
๐ธ Nami API Gateway โ Unified REST API for the Nami Army
One endpoint to rule them all.
Endpoints:
GET /api/status โ All services health
GET /api/services โ List all services
GET /api/agents โ Coding agent status
GET /api/bridge โ Nami Bridge status
POST /api/health/check โ Run health check
Run:
python3 /opt/nami-army/api_gateway.py --port 8700
"""
import asyncio
import json
import os
import subprocess
import sys
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional
# โโโ Dependencies check โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
try:
from fastapi import FastAPI, HTTPException, Query
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse, HTMLResponse
import uvicorn
except ImportError:
print("Installing dependencies...")
subprocess.run([sys.executable, "-m", "pip", "install", "fastapi", "uvicorn", "-q"])
from fastapi import FastAPI, HTTPException, Query
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse, HTMLResponse
import uvicorn
app = FastAPI(
title="๐ธ Nami API Gateway",
version="1.0.0",
description="Unified REST API for Nami Army services"
)
# CORS โ allow all (for dashboard, external tools)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["*"],
allow_headers=["*"],
)
NAMI_DIR = Path("/opt/nami-army")
SERVICES = {
"hermes-gateway": {"name": "Hermes Gateway", "emoji": "๐ง ", "port": 8644, "critical": True},
"nami-bridge": {"name": "Nami Bridge", "emoji": "๐", "port": 8888, "critical": True},
"nami-status-api": {"name": "Status API", "emoji": "๐ก", "port": 8699, "critical": False},
"laopatana-stat-lab": {"name": "LaoPatana", "emoji": "๐", "port": 3000, "critical": False},
"hanoi-stats-analyzer": {"name": "Hanoi Lottery", "emoji": "๐ฒ", "port": 3002, "critical": False},
"clipboardbypao": {"name": "ClipboardByPao", "emoji": "๐", "port": 3001, "critical": False},
"gold-signal-os": {"name": "Gold Signal", "emoji": "๐ฅ", "port": 8002, "critical": False},
"miroshark-backend": {"name": "MiroShark Backend", "emoji": "๐ฆ", "port": 5001, "critical": False},
"miroshark-frontend": {"name": "MiroShark Frontend","emoji": "๐ฆ", "port": 3003, "critical": False},
"miroshark-oracle": {"name": "MiroShark Oracle", "emoji": "๐ฎ", "port": 8003, "critical": False},
"nginx": {"name": "Nginx", "emoji": "๐", "critical": True},
"postgresql": {"name": "PostgreSQL", "emoji": "๐๏ธ", "critical": True},
}
def check_systemd(name: str) -> dict:
try:
r = subprocess.run(["systemctl", "is-active", name], capture_output=True, text=True, timeout=5)
return {"status": "up" if r.stdout.strip() == "active" else "down", "detail": r.stdout.strip()}
except:
return {"status": "down", "detail": "error"}
def get_system_metrics() -> dict:
"""Get system metrics: CPU, RAM, Disk, Uptime"""
import shutil
metrics = {}
# CPU โ use /proc/loadavg for simplicity
try:
with open('/proc/loadavg') as f:
load = f.read().split()[0]
metrics['cpu'] = f"{float(load):.1f}"
except:
metrics['cpu'] = "โ"
# RAM
try:
mem = {}
with open('/proc/meminfo') as f:
for line in f:
if 'MemTotal' in line: mem['total'] = int(line.split()[1])
if 'MemAvailable' in line: mem['avail'] = int(line.split()[1])
pct = (1 - mem['avail']/mem['total']) * 100
metrics['ram'] = f"{pct:.0f}"
except:
metrics['ram'] = "โ"
# Disk
try:
usage = shutil.disk_usage('/')
pct = (1 - usage.free/usage.total) * 100
metrics['disk'] = f"{pct:.0f}"
except:
metrics['disk'] = "โ"
# Uptime
try:
with open('/proc/uptime') as f:
secs = int(float(f.read().split()[0]))
days = secs // 86400
hours = (secs % 86400) // 3600
mins = (secs % 3600) // 60
parts = []
if days: parts.append(f"{days}d")
if hours: parts.append(f"{hours}h")
parts.append(f"{mins}m")
metrics['uptime'] = ' '.join(parts)
except:
metrics['uptime'] = "โ"
return metrics
def check_all_services() -> dict:
results = {}
now = datetime.now(timezone.utc).isoformat()
for svc_id, svc_def in SERVICES.items():
if "port" in svc_def and svc_def.get("type") != "systemd_only":
# Check both systemd and port
sysd = check_systemd(svc_id) if svc_id not in ("hermes-gateway",) else {"status": "up", "detail": "managed by hermes"}
else:
sysd = check_systemd(svc_id)
results[svc_id] = {
"name": svc_def["name"],
"emoji": svc_def["emoji"],
"status": sysd["status"],
"detail": sysd.get("detail", ""),
"critical": svc_def.get("critical", False),
"checked_at": now
}
return results
# โโโ API Endpoints โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
@app.get("/", response_class=HTMLResponse)
async def root():
"""Landing page"""
return """
๐ธ Nami API Gateway
"""
@app.get("/api/status")
async def api_status():
"""Get status of all services"""
results = check_all_services()
metrics = get_system_metrics()
up = sum(1 for s in results.values() if s["status"] == "up")
down = sum(1 for s in results.values() if s["status"] == "down")
critical_down = sum(1 for s in results.values() if s["status"] == "down" and s.get("critical"))
return {
"ok": down == 0,
"total": len(results),
"up": up,
"down": down,
"critical_down": critical_down,
"cpu": metrics["cpu"],
"ram": metrics["ram"],
"disk": metrics["disk"],
"uptime": metrics["uptime"],
"timestamp": datetime.now(timezone.utc).isoformat(),
"services": results,
"vps": "178.104.181.132",
"name": "๐ธ Nami Army"
}
@app.get("/api/services")
async def api_services():
"""List all services with endpoints"""
return {
"services": [
{"name": "๐ธ Nami Dashboard", "url": "http://ecosystem.178.104.181.132.nip.io"},
{"name": "๐ LaoPatana", "url": "http://lao.178.104.181.132.nip.io"},
{"name": "๐ฒ Hanoi Stats", "url": "http://hanoi.178.104.181.132.nip.io"},
{"name": "๐ ClipboardByPao", "url": "http://clipboardbypao.178.104.181.132.nip.io"},
{"name": "๐ฅ Gold Signal", "url": "http://goldsignalos.178.104.181.132.nip.io"},
{"name": "๐ฆ MiroShark", "url": "http://miroshark.178.104.181.132.nip.io"},
{"name": "๐ง Hermes Gateway", "url": "http://178.104.181.132:8644"},
{"name": "๐ Nami Bridge (WS)", "url": "ws://178.104.181.132:8888"},
{"name": "๐ก API Gateway", "url": "http://178.104.181.132:8700"},
]
}
@app.get("/api/agents")
async def api_agents():
"""Coding agent status"""
return {
"agents": [
{"name": "Claude Code", "emoji": "๐ต", "status": "active", "provider": "MaxplusAI Proxy", "version": "2.1.126"},
{"name": "OpenCode", "emoji": "๐ข", "status": "active", "provider": "OpenRouter", "model": "nemotron-120b:free"},
{"name": "thClaws", "emoji": "๐ฆ", "status": "active", "provider": "OpenRouter", "version": "0.7.4", "model": "nemotron-9b:free"},
{"name": "Nami Router", "emoji": "๐งญ", "status": "active", "version": "2.0"},
],
"router": {
"architecture": "deepseekโopenrouterโnvidiaโmaxplus",
"total_models": 371,
"free_models": ["nemotron-120b", "nemotron-nano-9b", "nemotron-30b-reasoning"]
}
}
@app.get("/api/bridge")
async def api_bridge():
"""Nami Bridge status"""
# Try connecting as agent
bridge_status = "up"
browser_status = "unknown"
try:
import websockets
async with websockets.connect("ws://127.0.0.1:8888") as ws:
await ws.send(json.dumps({"type": "hello", "role": "agent", "name": "API-Gateway"}))
ack_raw = await asyncio.wait_for(ws.recv(), timeout=3)
ack = json.loads(ack_raw)
# Ping to check browser
await ws.send(json.dumps({"type": "ping"}))
try:
pong_raw = await asyncio.wait_for(ws.recv(), timeout=3)
pong = json.loads(pong_raw)
browser_status = "connected" if pong.get("type") == "pong" else "error"
except:
browser_status = "disconnected"
except:
bridge_status = "down"
return {
"bridge": {
"status": bridge_status,
"port": 8888,
"protocol": "websocket",
"url": "ws://178.104.181.132:8888"
},
"browser": {
"status": browser_status,
"extension_version": "1.2.0",
"install_url": "/nami-extension-v1.2.0.zip"
}
}
@app.get("/api/health")
async def api_health():
"""Simple health check"""
return {"status": "ok", "service": "Nami API Gateway", "version": "1.0.0"}
@app.get("/api/backup")
async def api_backup():
"""Backup status โ reads from backup_state.json"""
state_file = Path("/opt/backup/backup_state.json")
if state_file.exists():
state = json.loads(state_file.read_text())
return {
"status": "ok",
"last_backup": state.get("last_backup", "never"),
"total_size_mb": state.get("total_size_mb", 0),
"databases": state.get("databases", 0),
"success_count": state.get("success_count", 0),
"failed": state.get("failed", []),
"kept_days": state.get("kept_days", 14),
"disk_free_gb": state.get("disk_free_gb", 0),
}
return {"status": "unknown", "last_backup": "never"}
@app.get("/health")
async def health():
"""Health check (for load balancers)"""
return {"status": "ok"}
# โโโ Main โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
def main():
import argparse
parser = argparse.ArgumentParser(description="๐ธ Nami API Gateway")
parser.add_argument("--port", type=int, default=8700, help="Port (default: 8700)")
parser.add_argument("--host", default="0.0.0.0", help="Host (default: 0.0.0.0)")
args = parser.parse_args()
print(f"๐ธ Nami API Gateway v1.0.0")
print(f" http://{args.host}:{args.port}")
print(f" Docs: http://{args.host}:{args.port}/docs")
uvicorn.run(app, host=args.host, port=args.port, log_level="info")
if __name__ == "__main__":
main()