# -*- coding: utf-8 -*- """Read-only OpenClaw CLI API routes — typed with Pydantic models.""" from __future__ import annotations from typing import Any from fastapi import APIRouter, Depends, HTTPException, Query from pydantic import BaseModel, Field from backend.services.openclaw_cli import OpenClawCliError, OpenClawCliService from shared.models.openclaw import OpenClawStatus router = APIRouter(prefix="/api/openclaw", tags=["openclaw"]) def get_openclaw_cli_service() -> OpenClawCliService: """Build the OpenClaw CLI service dependency.""" return OpenClawCliService() def _raise_cli_http_error(exc: OpenClawCliError) -> None: detail = { "message": str(exc), "command": exc.command, "exit_code": exc.exit_code, "stdout": exc.stdout, "stderr": exc.stderr, } status_code = 503 if exc.exit_code is None else 502 raise HTTPException(status_code=status_code, detail=detail) from exc # --------------------------------------------------------------------------- # Response wrappers # --------------------------------------------------------------------------- class StatusResponse(BaseModel): status: object class SessionsResponse(BaseModel): sessions: list[object] class SessionDetailResponse(BaseModel): session: object | None class SessionHistoryResponse(BaseModel): session_key: str session_id: str | None events: list[object] history: list[object] raw_text: str | None class CronResponse(BaseModel): cron: list[object] jobs: list[object] class ApprovalsResponse(BaseModel): approvals: list[object] pending: list[object] class AgentsResponse(BaseModel): agents: list[object] class SkillsResponse(BaseModel): workspace_dir: str managed_skills_dir: str skills: list[object] class ModelsResponse(BaseModel): models: list[object] class HooksResponse(BaseModel): workspace_dir: str managed_hooks_dir: str hooks: list[object] class PluginsResponse(BaseModel): workspace_dir: str plugins: list[object] diagnostics: list[object] class SecretsAuditResponse(BaseModel): version: int status: str findings: list[object] class SecurityAuditResponse2(BaseModel): report: object | None secret_diagnostics: list[str] class DaemonStatusResponse(BaseModel): service: object | None port: object | None rpc: object | None health: object | None class PairingListResponse2(BaseModel): channel: str requests: list[object] class QrCodeResponse2(BaseModel): setup_code: str gateway_url: str auth: str url_source: str class UpdateStatusResponse2(BaseModel): update: object | None channel: object | None class ModelAliasesResponse(BaseModel): aliases: dict[str, str] class ModelFallbacksResponse(BaseModel): key: str label: str items: list[object] class SkillUpdateResponse(BaseModel): ok: bool slug: str version: str error: str | None class ModelsStatusResponse(BaseModel): configPath: str | None = None agentId: str | None = None agentDir: str | None = None defaultModel: str | None = None resolvedDefault: str | None = None fallbacks: list[str] = Field(default_factory=list) imageModel: str | None = None imageFallbacks: list[str] = Field(default_factory=list) aliases: dict[str, str] = Field(default_factory=dict) allowed: list[str] = Field(default_factory=list) auth: dict[str, Any] = Field(default_factory=dict) class ChannelsStatusResponse(BaseModel): reachable: bool | None = None channelAccounts: dict[str, Any] = Field(default_factory=dict) channels: list[str] = Field(default_factory=list) issues: list[dict[str, Any]] = Field(default_factory=list) class ChannelsListResponse(BaseModel): chat: dict[str, list[str]] = Field(default_factory=dict) auth: list[dict[str, Any]] = Field(default_factory=list) usage: dict[str, Any] | None = None class HookInfoResponse(BaseModel): name: str | None = None description: str | None = None source: str | None = None pluginId: str | None = None filePath: str | None = None handlerPath: str | None = None hookKey: str | None = None emoji: str | None = None homepage: str | None = None events: list[str] = Field(default_factory=list) enabledByConfig: bool | None = None loadable: bool | None = None requirementsSatisfied: bool | None = None requirements: dict[str, Any] = Field(default_factory=dict) error: str | None = None raw: str | None = None class HooksCheckResponse(BaseModel): workspace_dir: str = "" managed_hooks_dir: str = "" hooks: list[dict[str, Any]] = Field(default_factory=list) eligible: bool | None = None verbose: bool | None = None class PluginInspectEntry(BaseModel): plugin: dict[str, Any] = Field(default_factory=dict) shape: str | None = None capabilityMode: str | None = None capabilityCount: int = 0 capabilities: list[dict[str, Any]] = Field(default_factory=list) typedHooks: list[dict[str, Any]] = Field(default_factory=list) customHooks: list[dict[str, Any]] = Field(default_factory=list) tools: list[dict[str, Any]] = Field(default_factory=list) commands: list[str] = Field(default_factory=list) cliCommands: list[str] = Field(default_factory=list) services: list[str] = Field(default_factory=list) gatewayMethods: list[str] = Field(default_factory=list) mcpServers: list[dict[str, Any]] = Field(default_factory=list) lspServers: list[dict[str, Any]] = Field(default_factory=list) httpRouteCount: int = 0 bundleCapabilities: list[str] = Field(default_factory=list) class PluginsInspectResponse(BaseModel): inspect: list[dict[str, Any]] = Field(default_factory=list) class AgentBindingItem(BaseModel): agentId: str match: dict[str, Any] description: str class AgentsBindingsResponse(BaseModel): bindings: list[AgentBindingItem] # --------------------------------------------------------------------------- # Routes — use typed model methods and return Pydantic models directly # --------------------------------------------------------------------------- @router.get("/status") async def api_openclaw_status( service: OpenClawCliService = Depends(get_openclaw_cli_service), ) -> OpenClawStatus: """Read `openclaw status --json` and return a typed model.""" try: return service.status_model() except OpenClawCliError as exc: _raise_cli_http_error(exc) @router.get("/sessions") async def api_openclaw_sessions( service: OpenClawCliService = Depends(get_openclaw_cli_service), ) -> SessionsResponse: """Read `openclaw sessions --json` and return a typed SessionsList.""" try: result = service.list_sessions_model() return SessionsResponse(sessions=result.sessions) except OpenClawCliError as exc: _raise_cli_http_error(exc) @router.get("/sessions/{session_key:path}/history") async def api_openclaw_session_history( session_key: str, limit: int = Query(20, ge=1, le=200), service: OpenClawCliService = Depends(get_openclaw_cli_service), ) -> SessionHistoryResponse: """Read session history and return a typed SessionHistory.""" try: result = service.get_session_history_model(session_key, limit=limit) return SessionHistoryResponse( session_key=result.session_key, session_id=result.session_id, events=result.events, history=result.events, # alias for compat raw_text=result.raw_text, ) except OpenClawCliError as exc: _raise_cli_http_error(exc) @router.get("/sessions/{session_key:path}") async def api_openclaw_session_detail( session_key: str, service: OpenClawCliService = Depends(get_openclaw_cli_service), ) -> SessionDetailResponse: """Resolve a single session and return it as a typed model.""" try: session = service.get_session_model(session_key) return SessionDetailResponse(session=session) except KeyError as exc: raise HTTPException(status_code=404, detail=f"session '{session_key}' not found") from exc except OpenClawCliError as exc: _raise_cli_http_error(exc) @router.get("/cron") async def api_openclaw_cron( service: OpenClawCliService = Depends(get_openclaw_cli_service), ) -> CronResponse: """Read `openclaw cron list --json` and return a typed CronList.""" try: result = service.list_cron_jobs_model() return CronResponse(cron=list(result.cron), jobs=list(result.jobs)) except OpenClawCliError as exc: _raise_cli_http_error(exc) @router.get("/approvals") async def api_openclaw_approvals( service: OpenClawCliService = Depends(get_openclaw_cli_service), ) -> ApprovalsResponse: """Read `openclaw approvals get --json` and return a typed ApprovalsList.""" try: result = service.list_approvals_model() return ApprovalsResponse( approvals=list(result.approvals), pending=list(result.pending), ) except OpenClawCliError as exc: _raise_cli_http_error(exc) @router.get("/agents") async def api_openclaw_agents( service: OpenClawCliService = Depends(get_openclaw_cli_service), ) -> AgentsResponse: """Read `openclaw agents list --json` and return a typed AgentsList.""" try: result = service.list_agents_model() return AgentsResponse(agents=list(result.agents)) except OpenClawCliError as exc: _raise_cli_http_error(exc) @router.get("/agents/presence") async def api_openclaw_agents_presence( service: OpenClawCliService = Depends(get_openclaw_cli_service), ) -> dict[str, Any]: """Read runtime session presence for all agents from session files.""" result = service.agents_presence() return result # --------------------------------------------------------------------------- # Write agents routes # --------------------------------------------------------------------------- class AgentAddResponse(BaseModel): agentId: str name: str workspace: str agentDir: str model: str | None = None bindings: dict[str, Any] = Field(default_factory=dict) class AgentDeleteResponse(BaseModel): agentId: str workspace: str agentDir: str sessionsDir: str removedBindings: list[str] = Field(default_factory=list) removedAllow: list[str] = Field(default_factory=list) class AgentBindResponse(BaseModel): agentId: str added: list[str] = Field(default_factory=list) updated: list[str] = Field(default_factory=list) skipped: list[str] = Field(default_factory=list) conflicts: list[str] = Field(default_factory=list) class AgentUnbindResponse(BaseModel): agentId: str removed: list[str] = Field(default_factory=list) missing: list[str] = Field(default_factory=list) conflicts: list[str] = Field(default_factory=list) class AgentIdentityResponse(BaseModel): agentId: str identity: dict[str, Any] = Field(default_factory=dict) workspace: str | None = None identityFile: str | None = None @router.post("/agents/add") async def api_openclaw_agents_add( name: str, *, workspace: str | None = None, model: str | None = None, agent_dir: str | None = None, bind: list[str] | None = None, non_interactive: bool = False, service: OpenClawCliService = Depends(get_openclaw_cli_service), ) -> AgentAddResponse: """Run `openclaw agents add ` and return JSON result.""" try: result = service.agents_add( name, workspace=workspace, model=model, agent_dir=agent_dir, bind=bind, non_interactive=non_interactive, ) return AgentAddResponse.model_validate(result, strict=False) except OpenClawCliError as exc: _raise_cli_http_error(exc) @router.post("/agents/delete/{id}") async def api_openclaw_agents_delete( id: str, force: bool = False, service: OpenClawCliService = Depends(get_openclaw_cli_service), ) -> AgentDeleteResponse: """Run `openclaw agents delete [--force]` and return JSON result.""" try: result = service.agents_delete(id, force=force) return AgentDeleteResponse.model_validate(result, strict=False) except OpenClawCliError as exc: _raise_cli_http_error(exc) @router.post("/agents/bind") async def api_openclaw_agents_bind( *, agent: str | None = None, bind: list[str] | None = None, service: OpenClawCliService = Depends(get_openclaw_cli_service), ) -> AgentBindResponse: """Run `openclaw agents bind [--agent ] [--bind ]` and return JSON result.""" try: result = service.agents_bind(agent=agent, bind=bind) return AgentBindResponse.model_validate(result, strict=False) except OpenClawCliError as exc: _raise_cli_http_error(exc) @router.post("/agents/unbind") async def api_openclaw_agents_unbind( *, agent: str | None = None, bind: list[str] | None = None, all: bool = False, service: OpenClawCliService = Depends(get_openclaw_cli_service), ) -> AgentUnbindResponse: """Run `openclaw agents unbind [--agent ] [--bind ] [--all]` and return JSON result.""" try: result = service.agents_unbind(agent=agent, bind=bind, all=all) return AgentUnbindResponse.model_validate(result, strict=False) except OpenClawCliError as exc: _raise_cli_http_error(exc) @router.post("/agents/set-identity") async def api_openclaw_agents_set_identity( *, agent: str | None = None, workspace: str | None = None, identity_file: str | None = None, name: str | None = None, emoji: str | None = None, theme: str | None = None, avatar: str | None = None, from_identity: bool = False, service: OpenClawCliService = Depends(get_openclaw_cli_service), ) -> AgentIdentityResponse: """Run `openclaw agents set-identity` and return JSON result.""" try: result = service.agents_set_identity( agent=agent, workspace=workspace, identity_file=identity_file, name=name, emoji=emoji, theme=theme, avatar=avatar, from_identity=from_identity, ) return AgentIdentityResponse.model_validate(result, strict=False) except OpenClawCliError as exc: _raise_cli_http_error(exc) @router.get("/skills") async def api_openclaw_skills( service: OpenClawCliService = Depends(get_openclaw_cli_service), ) -> SkillsResponse: """Read `openclaw skills list --json` and return a typed SkillStatusReport.""" try: result = service.list_skills_model() return SkillsResponse( workspace_dir=result.workspace_dir, managed_skills_dir=result.managed_skills_dir, skills=list(result.skills), ) except OpenClawCliError as exc: _raise_cli_http_error(exc) @router.get("/models") async def api_openclaw_models( service: OpenClawCliService = Depends(get_openclaw_cli_service), ) -> ModelsResponse: """Read `openclaw models list --json` and return a typed ModelsList.""" try: result = service.list_models_model() return ModelsResponse(models=list(result.models)) except OpenClawCliError as exc: _raise_cli_http_error(exc) @router.get("/hooks") async def api_openclaw_hooks( service: OpenClawCliService = Depends(get_openclaw_cli_service), ) -> HooksResponse: try: result = service.list_hooks_model() return HooksResponse( workspace_dir=result.workspace_dir, managed_hooks_dir=result.managed_hooks_dir, hooks=list(result.hooks), ) except OpenClawCliError as exc: _raise_cli_http_error(exc) @router.get("/plugins") async def api_openclaw_plugins( service: OpenClawCliService = Depends(get_openclaw_cli_service), ) -> PluginsResponse: try: result = service.list_plugins_model() return PluginsResponse( workspace_dir=result.workspace_dir, plugins=list(result.plugins), diagnostics=list(result.diagnostics), ) except OpenClawCliError as exc: _raise_cli_http_error(exc) @router.get("/secrets-audit") async def api_openclaw_secrets_audit( service: OpenClawCliService = Depends(get_openclaw_cli_service), ) -> SecretsAuditResponse: try: result = service.secrets_audit_model() return SecretsAuditResponse( version=result.version, status=result.status, findings=list(result.findings), ) except OpenClawCliError as exc: _raise_cli_http_error(exc) @router.get("/security-audit") async def api_openclaw_security_audit( service: OpenClawCliService = Depends(get_openclaw_cli_service), ) -> SecurityAuditResponse2: try: result = service.security_audit_model() return SecurityAuditResponse2( report=result.report.model_dump() if result.report else None, secret_diagnostics=list(result.secret_diagnostics), ) except OpenClawCliError as exc: _raise_cli_http_error(exc) @router.get("/daemon-status") async def api_openclaw_daemon_status( service: OpenClawCliService = Depends(get_openclaw_cli_service), ) -> DaemonStatusResponse: try: result = service.daemon_status_model() return DaemonStatusResponse( service=result.service.model_dump() if result.service else None, port=result.port.model_dump() if result.port else None, rpc=result.rpc.model_dump() if result.rpc else None, health=result.health.model_dump() if result.health else None, ) except OpenClawCliError as exc: _raise_cli_http_error(exc) @router.get("/pairing") async def api_openclaw_pairing( service: OpenClawCliService = Depends(get_openclaw_cli_service), ) -> PairingListResponse2: try: result = service.pairing_list_model() return PairingListResponse2( channel=result.channel, requests=list(result.requests), ) except OpenClawCliError as exc: _raise_cli_http_error(exc) @router.get("/qr") async def api_openclaw_qr( service: OpenClawCliService = Depends(get_openclaw_cli_service), ) -> QrCodeResponse2: try: result = service.qr_code_model() return QrCodeResponse2( setup_code=result.setup_code, gateway_url=result.gateway_url, auth=result.auth, url_source=result.url_source, ) except OpenClawCliError as exc: _raise_cli_http_error(exc) @router.get("/update-status") async def api_openclaw_update_status( service: OpenClawCliService = Depends(get_openclaw_cli_service), ) -> UpdateStatusResponse2: try: result = service.update_status_model() return UpdateStatusResponse2( update=result.update.model_dump() if result.update else None, channel=result.channel.model_dump() if result.channel else None, ) except OpenClawCliError as exc: _raise_cli_http_error(exc) @router.get("/models-aliases") async def api_openclaw_models_aliases( service: OpenClawCliService = Depends(get_openclaw_cli_service), ) -> ModelAliasesResponse: try: result = service.list_model_aliases_model() return ModelAliasesResponse(aliases=result.aliases) except OpenClawCliError as exc: _raise_cli_http_error(exc) @router.get("/models-fallbacks") async def api_openclaw_models_fallbacks( service: OpenClawCliService = Depends(get_openclaw_cli_service), ) -> ModelFallbacksResponse: try: result = service.list_model_fallbacks_model() return ModelFallbacksResponse( key=result.key, label=result.label, items=list(result.items), ) except OpenClawCliError as exc: _raise_cli_http_error(exc) @router.get("/models-image-fallbacks") async def api_openclaw_models_image_fallbacks( service: OpenClawCliService = Depends(get_openclaw_cli_service), ) -> ModelFallbacksResponse: try: result = service.list_model_image_fallbacks_model() return ModelFallbacksResponse( key=result.key, label=result.label, items=list(result.items), ) except OpenClawCliError as exc: _raise_cli_http_error(exc) @router.get("/skill-update") async def api_openclaw_skill_update( slug: str | None = None, all: bool = False, service: OpenClawCliService = Depends(get_openclaw_cli_service), ) -> SkillUpdateResponse: try: result = service.skill_update_model(slug=slug, all=all) return SkillUpdateResponse( ok=result.ok, slug=result.slug, version=result.version, error=result.error, ) except OpenClawCliError as exc: _raise_cli_http_error(exc) @router.get("/models-status") async def api_openclaw_models_status( probe: bool = False, service: OpenClawCliService = Depends(get_openclaw_cli_service), ) -> ModelsStatusResponse: """Read `openclaw models status --json [--probe]` and return a typed dict.""" try: result = service.models_status_model(probe=probe) return ModelsStatusResponse.model_validate(result, strict=False) except OpenClawCliError as exc: _raise_cli_http_error(exc) @router.get("/channels-status") async def api_openclaw_channels_status( probe: bool = False, service: OpenClawCliService = Depends(get_openclaw_cli_service), ) -> ChannelsStatusResponse: """Read `openclaw channels status --json [--probe]` and return a typed dict.""" try: result = service.channels_status_model(probe=probe) return ChannelsStatusResponse.model_validate(result, strict=False) except OpenClawCliError as exc: _raise_cli_http_error(exc) @router.get("/channels-list") async def api_openclaw_channels_list( service: OpenClawCliService = Depends(get_openclaw_cli_service), ) -> ChannelsListResponse: """Read `openclaw channels list --json` and return a typed dict.""" try: result = service.channels_list_model() return ChannelsListResponse.model_validate(result, strict=False) except OpenClawCliError as exc: _raise_cli_http_error(exc) @router.get("/hooks/info/{name}") async def api_openclaw_hook_info( name: str, service: OpenClawCliService = Depends(get_openclaw_cli_service), ) -> HookInfoResponse: """Read `openclaw hooks info --json` and return a typed dict.""" try: result = service.hook_info_model(name) return HookInfoResponse.model_validate(result, strict=False) except OpenClawCliError as exc: _raise_cli_http_error(exc) @router.get("/hooks/check") async def api_openclaw_hooks_check( service: OpenClawCliService = Depends(get_openclaw_cli_service), ) -> HooksCheckResponse: """Read `openclaw hooks check --json` and return a typed dict.""" try: result = service.hooks_check_model() return HooksCheckResponse.model_validate(result, strict=False) except OpenClawCliError as exc: _raise_cli_http_error(exc) @router.get("/plugins-inspect") async def api_openclaw_plugins_inspect( plugin_id: str | None = None, all: bool = False, service: OpenClawCliService = Depends(get_openclaw_cli_service), ) -> PluginsInspectResponse: """Read `openclaw plugins inspect --json [--all]` and return a typed dict.""" try: result = service.plugins_inspect_model(plugin_id=plugin_id, all=all) inspect = result if isinstance(result, list) else result.get("inspect", []) return PluginsInspectResponse(inspect=inspect) except OpenClawCliError as exc: _raise_cli_http_error(exc) class AgentBindingItem(BaseModel): agentId: str match: dict[str, Any] description: str class AgentsBindingsResponse(BaseModel): bindings: list[AgentBindingItem] @router.get("/agents-bindings") async def api_openclaw_agents_bindings( agent: str | None = None, service: OpenClawCliService = Depends(get_openclaw_cli_service), ) -> AgentsBindingsResponse: """Read `openclaw agents bindings --json [--agent ]` and return bindings list.""" try: result = service.agents_bindings_model(agent=agent) bindings = result if isinstance(result, list) else [] return AgentsBindingsResponse(bindings=bindings) except OpenClawCliError as exc: _raise_cli_http_error(exc) @router.get("/gateway-status") async def api_openclaw_gateway_status( url: str | None = None, token: str | None = None, service: OpenClawCliService = Depends(get_openclaw_cli_service), ) -> dict[str, Any]: """Read `openclaw gateway status --json [--url ] [--token ]`. Returns full gateway probe result.""" try: result = service.gateway_status(url=url, token=token) return result except OpenClawCliError as exc: _raise_cli_http_error(exc) @router.get("/memory-status") async def api_openclaw_memory_status( agent: str | None = None, deep: bool = False, service: OpenClawCliService = Depends(get_openclaw_cli_service), ) -> list[dict[str, Any]]: """Read `openclaw memory status --json [--agent ] [--deep]`. Returns array of per-agent memory status.""" try: result = service.memory_status(agent=agent, deep=deep) return result if isinstance(result, list) else [] except OpenClawCliError as exc: _raise_cli_http_error(exc) class WorkspaceFilesResponse(BaseModel): workspace: str files: list[dict[str, Any]] error: str | None = None @router.get("/workspace-files") async def api_openclaw_workspace_files( workspace: str = Query(..., description="Path to the agent workspace directory"), service: OpenClawCliService = Depends(get_openclaw_cli_service), ) -> WorkspaceFilesResponse: """List .md files in an OpenClaw agent workspace with their content previews.""" result = service.list_workspace_files(workspace) return WorkspaceFilesResponse.model_validate(result, strict=False)