392 lines
16 KiB
Python
392 lines
16 KiB
Python
"""
|
|
🐙 OpenClaw Service — Status, Configuration, Agents, Skills, Logs & Models.
|
|
|
|
Reads data from the OpenClaw home directory (FOXY_HOME) and exposes it
|
|
through a set of REST endpoints for the dashboard frontend.
|
|
"""
|
|
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
import socket
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any, Optional
|
|
|
|
from fastapi import APIRouter, Query
|
|
|
|
from app.config import settings
|
|
|
|
log = logging.getLogger("foxy.api.openclaw")
|
|
router = APIRouter(prefix="/api/openclaw", tags=["openclaw"])
|
|
|
|
# Secrets / tokens to mask in config output
|
|
_SECRET_PATTERNS = re.compile(
|
|
r"(token|key|secret|password|apiKey|api_key)", re.IGNORECASE
|
|
)
|
|
|
|
|
|
def _home() -> Path:
|
|
"""Resolve the OpenClaw home directory."""
|
|
return Path(settings.FOXY_HOME)
|
|
|
|
|
|
def _mask_secrets(obj: Any, depth: int = 0) -> Any:
|
|
"""Recursively mask values whose keys look like secrets."""
|
|
if depth > 20:
|
|
return obj
|
|
if isinstance(obj, dict):
|
|
masked = {}
|
|
for k, v in obj.items():
|
|
if isinstance(v, str) and _SECRET_PATTERNS.search(k) and v:
|
|
masked[k] = "••••••••"
|
|
else:
|
|
masked[k] = _mask_secrets(v, depth + 1)
|
|
return masked
|
|
if isinstance(obj, list):
|
|
return [_mask_secrets(item, depth + 1) for item in obj]
|
|
return obj
|
|
|
|
|
|
def _port_open(port: int, host: str = "127.0.0.1", timeout: float = 1.0) -> bool:
|
|
"""Check if a TCP port is open."""
|
|
try:
|
|
with socket.create_connection((host, port), timeout=timeout):
|
|
return True
|
|
except (OSError, ConnectionRefusedError, TimeoutError):
|
|
return False
|
|
|
|
|
|
def _read_json(path: Path) -> Optional[dict]:
|
|
"""Read and parse a JSON file, return None on failure."""
|
|
try:
|
|
return json.loads(path.read_text(encoding="utf-8"))
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _dir_tree(root: Path, max_depth: int = 3, _depth: int = 0) -> list[dict]:
|
|
"""Build a lightweight directory tree."""
|
|
entries: list[dict] = []
|
|
if not root.is_dir() or _depth > max_depth:
|
|
return entries
|
|
try:
|
|
for child in sorted(root.iterdir()):
|
|
name = child.name
|
|
if name.startswith(".") and name not in (".openclaw",):
|
|
continue
|
|
if name in ("node_modules", "__pycache__", ".git"):
|
|
continue
|
|
entry: dict = {"name": name, "path": str(child.relative_to(_home()))}
|
|
if child.is_dir():
|
|
entry["type"] = "directory"
|
|
entry["children"] = _dir_tree(child, max_depth, _depth + 1)
|
|
else:
|
|
entry["type"] = "file"
|
|
try:
|
|
entry["size"] = child.stat().st_size
|
|
except OSError:
|
|
entry["size"] = 0
|
|
entries.append(entry)
|
|
except PermissionError:
|
|
pass
|
|
return entries
|
|
|
|
|
|
# ═════════════════════════════════════════════════════════════════════════════════
|
|
# 1. Status
|
|
# ═════════════════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
@router.get("/status")
|
|
async def get_status():
|
|
"""Gateway health, process info, mode."""
|
|
home = _home()
|
|
openclaw_type = os.environ.get("OPENCLAW_TYPE", "shared")
|
|
|
|
# Check gateway port
|
|
gateway_up = await asyncio.to_thread(_port_open, 18789)
|
|
|
|
# Try reading config for extra info
|
|
config_data = await asyncio.to_thread(_read_json, home / "openclaw.json")
|
|
gateway_cfg = {}
|
|
if config_data:
|
|
gateway_cfg = config_data.get("gateway", {})
|
|
|
|
# Count agents, skills
|
|
agents_dir = home / "agents"
|
|
skills_dir = home / "skills"
|
|
agent_count = len(list(agents_dir.iterdir())) if agents_dir.is_dir() else 0
|
|
skill_count = len(list(skills_dir.iterdir())) if skills_dir.is_dir() else 0
|
|
|
|
# Gateway log existence
|
|
log_file = home / "logs" / "gateway.log"
|
|
log_size = 0
|
|
if log_file.is_file():
|
|
try:
|
|
log_size = log_file.stat().st_size
|
|
except OSError:
|
|
pass
|
|
|
|
return {
|
|
"gateway_online": gateway_up,
|
|
"openclaw_type": openclaw_type,
|
|
"foxy_home": str(home),
|
|
"gateway_bind": gateway_cfg.get("bind", "unknown"),
|
|
"gateway_mode": gateway_cfg.get("mode", "unknown"),
|
|
"gateway_port": 18789,
|
|
"agent_count": agent_count,
|
|
"skill_count": skill_count,
|
|
"log_file_size": log_size,
|
|
"config_exists": (home / "openclaw.json").is_file(),
|
|
"timestamp": datetime.now(timezone.utc).isoformat(),
|
|
}
|
|
|
|
|
|
# ═════════════════════════════════════════════════════════════════════════════════
|
|
# 2. Configuration
|
|
# ═════════════════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
@router.get("/config")
|
|
async def get_openclaw_config():
|
|
"""Read openclaw.json with secrets masked."""
|
|
config_path = _home() / "openclaw.json"
|
|
data = await asyncio.to_thread(_read_json, config_path)
|
|
if data is None:
|
|
return {"error": "openclaw.json not found or unreadable", "config": None}
|
|
return {"config": _mask_secrets(data)}
|
|
|
|
|
|
# ═════════════════════════════════════════════════════════════════════════════════
|
|
# 3. Agents
|
|
# ═════════════════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
@router.get("/agents")
|
|
async def list_openclaw_agents():
|
|
"""List agent definitions from the agents/ directory."""
|
|
agents_dir = _home() / "agents"
|
|
if not agents_dir.is_dir():
|
|
return {"agents": [], "error": "agents/ directory not found"}
|
|
|
|
agents = []
|
|
for entry in sorted(agents_dir.iterdir()):
|
|
agent: dict[str, Any] = {"name": entry.name}
|
|
|
|
if entry.is_dir():
|
|
# Directory-based agent — look for config files
|
|
agent["type"] = "directory"
|
|
config_files = list(entry.glob("*.json")) + list(entry.glob("*.yaml")) + list(entry.glob("*.yml"))
|
|
agent["config_files"] = [f.name for f in config_files]
|
|
|
|
# Try to read agent config
|
|
for cfg_file in config_files:
|
|
cfg = _read_json(cfg_file) if cfg_file.suffix == ".json" else None
|
|
if cfg:
|
|
agent["config"] = _mask_secrets(cfg)
|
|
agent["model"] = cfg.get("model", cfg.get("modelId", None))
|
|
agent["system_prompt"] = cfg.get("systemPrompt", cfg.get("system_prompt", cfg.get("instructions", None)))
|
|
if agent["system_prompt"] and len(agent["system_prompt"]) > 300:
|
|
agent["system_prompt_preview"] = agent["system_prompt"][:300] + "…"
|
|
break
|
|
|
|
# Count files
|
|
try:
|
|
all_files = list(entry.rglob("*"))
|
|
agent["file_count"] = len([f for f in all_files if f.is_file()])
|
|
except Exception:
|
|
agent["file_count"] = 0
|
|
|
|
elif entry.is_file() and entry.suffix == ".json":
|
|
# File-based agent definition
|
|
agent["type"] = "file"
|
|
cfg = _read_json(entry)
|
|
if cfg:
|
|
agent["config"] = _mask_secrets(cfg)
|
|
agent["model"] = cfg.get("model", cfg.get("modelId", None))
|
|
|
|
agents.append(agent)
|
|
|
|
return {"agents": agents, "count": len(agents)}
|
|
|
|
|
|
# ═════════════════════════════════════════════════════════════════════════════════
|
|
# 4. Skills
|
|
# ═════════════════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
@router.get("/skills")
|
|
async def list_openclaw_skills():
|
|
"""List available skills from the skills/ directory."""
|
|
# Skills can be in multiple locations
|
|
possible_dirs = [
|
|
_home() / "skills",
|
|
_home() / "workspace" / "skills",
|
|
]
|
|
|
|
skills = []
|
|
for skills_dir in possible_dirs:
|
|
if not skills_dir.is_dir():
|
|
continue
|
|
for entry in sorted(skills_dir.iterdir()):
|
|
skill: dict[str, Any] = {
|
|
"name": entry.name,
|
|
"source": str(skills_dir.relative_to(_home())),
|
|
}
|
|
|
|
if entry.is_dir():
|
|
skill["type"] = "directory"
|
|
# Look for SKILL.md or README.md
|
|
for doc_name in ("SKILL.md", "README.md", "skill.md", "readme.md"):
|
|
doc = entry / doc_name
|
|
if doc.is_file():
|
|
try:
|
|
content = doc.read_text(encoding="utf-8", errors="replace")
|
|
# Extract description from YAML frontmatter or first lines
|
|
skill["description"] = content[:500]
|
|
# Try extracting YAML description
|
|
if content.startswith("---"):
|
|
end = content.find("---", 3)
|
|
if end > 0:
|
|
frontmatter = content[3:end].strip()
|
|
for line in frontmatter.split("\n"):
|
|
if line.startswith("description:"):
|
|
skill["description_short"] = line.split(":", 1)[1].strip().strip("'\"")
|
|
break
|
|
except Exception:
|
|
pass
|
|
break
|
|
|
|
# Count files
|
|
try:
|
|
all_files = list(entry.rglob("*"))
|
|
skill["file_count"] = len([f for f in all_files if f.is_file()])
|
|
skill["subdirs"] = [d.name for d in entry.iterdir() if d.is_dir()]
|
|
except Exception:
|
|
skill["file_count"] = 0
|
|
|
|
elif entry.is_file():
|
|
skill["type"] = "file"
|
|
try:
|
|
skill["size"] = entry.stat().st_size
|
|
except OSError:
|
|
skill["size"] = 0
|
|
|
|
skills.append(skill)
|
|
|
|
return {"skills": skills, "count": len(skills)}
|
|
|
|
|
|
# ═════════════════════════════════════════════════════════════════════════════════
|
|
# 5. Logs
|
|
# ═════════════════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
@router.get("/logs")
|
|
async def get_gateway_logs(
|
|
lines: int = Query(200, ge=10, le=2000),
|
|
level: Optional[str] = Query(None, description="Filter by log level: INFO, WARNING, ERROR"),
|
|
):
|
|
"""Read the last N lines of the gateway log."""
|
|
log_file = _home() / "logs" / "gateway.log"
|
|
if not log_file.is_file():
|
|
# Try alternate locations
|
|
alt = _home() / "gateway.log"
|
|
if alt.is_file():
|
|
log_file = alt
|
|
else:
|
|
return {"lines": [], "error": "Gateway log file not found", "log_path": str(log_file)}
|
|
|
|
try:
|
|
content = await asyncio.to_thread(
|
|
log_file.read_text, encoding="utf-8", errors="replace"
|
|
)
|
|
all_lines = content.strip().split("\n")
|
|
|
|
# Take last N lines
|
|
result_lines = all_lines[-lines:]
|
|
|
|
# Filter by level if requested
|
|
if level:
|
|
level_upper = level.upper()
|
|
result_lines = [l for l in result_lines if level_upper in l.upper()]
|
|
|
|
return {
|
|
"lines": result_lines,
|
|
"total_lines": len(all_lines),
|
|
"showing": len(result_lines),
|
|
"log_path": str(log_file),
|
|
}
|
|
except Exception as e:
|
|
return {"lines": [], "error": str(e), "log_path": str(log_file)}
|
|
|
|
|
|
# ═════════════════════════════════════════════════════════════════════════════════
|
|
# 6. Models & Providers
|
|
# ═════════════════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
@router.get("/models")
|
|
async def list_models():
|
|
"""Extract configured models and providers from openclaw.json."""
|
|
config_path = _home() / "openclaw.json"
|
|
data = await asyncio.to_thread(_read_json, config_path)
|
|
if data is None:
|
|
return {"models": [], "providers": [], "error": "openclaw.json not found"}
|
|
|
|
# Extract providers
|
|
providers = []
|
|
providers_cfg = data.get("mcpServers", data.get("providers", data.get("llm", {})))
|
|
if isinstance(providers_cfg, dict):
|
|
for name, cfg in providers_cfg.items():
|
|
provider: dict[str, Any] = {
|
|
"name": name,
|
|
"type": cfg.get("type", cfg.get("provider", "unknown")) if isinstance(cfg, dict) else "unknown",
|
|
}
|
|
if isinstance(cfg, dict):
|
|
provider["enabled"] = cfg.get("enabled", True)
|
|
provider["model"] = cfg.get("model", cfg.get("modelId", None))
|
|
provider["endpoint"] = cfg.get("endpoint", cfg.get("baseUrl", cfg.get("url", None)))
|
|
providers.append(provider)
|
|
|
|
# Extract models list if present
|
|
models = []
|
|
models_cfg = data.get("models", [])
|
|
if isinstance(models_cfg, list):
|
|
for m in models_cfg:
|
|
if isinstance(m, dict):
|
|
models.append(_mask_secrets(m))
|
|
elif isinstance(m, str):
|
|
models.append({"name": m})
|
|
elif isinstance(models_cfg, dict):
|
|
for name, cfg in models_cfg.items():
|
|
model_entry = {"name": name}
|
|
if isinstance(cfg, dict):
|
|
model_entry.update(_mask_secrets(cfg))
|
|
models.append(model_entry)
|
|
|
|
return {
|
|
"models": models,
|
|
"providers": _mask_secrets(providers),
|
|
"raw_keys": list(data.keys()),
|
|
}
|
|
|
|
|
|
# ═════════════════════════════════════════════════════════════════════════════════
|
|
# 7. Filesystem Tree
|
|
# ═════════════════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
@router.get("/filesystem")
|
|
async def get_filesystem():
|
|
"""Get a directory tree of the OpenClaw home directory."""
|
|
home = _home()
|
|
if not home.is_dir():
|
|
return {"tree": [], "error": f"FOXY_HOME ({home}) does not exist"}
|
|
|
|
tree = await asyncio.to_thread(_dir_tree, home)
|
|
return {"root": str(home), "tree": tree}
|