# retoor import urllib.request import urllib.error import json import os class NinjaError(Exception): pass class Ninja: def __init__(self, base_url=None, ninja_user=None, ninja_session=None): self.base_url = base_url or os.environ.get("NINJAURL", "http://localhost:13370") self.ninja_user = ninja_user or os.environ.get("NINJA_USER", "") self.ninja_session = ninja_session or os.environ.get("NINJA_SESSION", "") def _request(self, method, path, body=None): url = self.base_url.rstrip("/") + path headers = {"Content-Type": "application/json"} if self.ninja_user: headers["X-Ninja-User"] = self.ninja_user if self.ninja_session: headers["X-Ninja-Session"] = self.ninja_session data = json.dumps(body).encode("utf-8") if body is not None else None req = urllib.request.Request(url, data=data, headers=headers, method=method) try: with urllib.request.urlopen(req) as resp: return json.loads(resp.read().decode("utf-8")) except urllib.error.HTTPError as e: try: detail = json.loads(e.read().decode("utf-8")) except Exception: detail = {"error": str(e)} raise NinjaError(detail) except urllib.error.URLError as e: raise NinjaError({"error": str(e.reason)}) def health(self): return self._request("GET", "/api/health") def list_tools(self): return self._request("GET", "/api/tools") def stats(self): return self._request("GET", "/api/stats") def stats_by_tool(self, tool_name): return self._request("GET", "/api/stats/" + tool_name) def stats_by_user(self, user): return self._request("GET", "/api/stats/user/" + user) def audit(self): return self._request("GET", "/api/audit") def audit_by_tool(self, tool_name): return self._request("GET", "/api/audit/" + tool_name) def peers(self): return self._request("GET", "/api/peers") def register_peer(self, address, ninja_user=None): body = {"address": address} if ninja_user is not None: body["ninja_user"] = ninja_user return self._request("POST", "/api/peers/register", body) def main_peer(self): return self._request("GET", "/api/peers/main") def heartbeat(self, address): return self._request("POST", "/api/peers/heartbeat", {"address": address}) def execute_tool(self, tool_name, **kwargs): params = {k: v for k, v in kwargs.items() if v is not None} return self._request("POST", "/api/tools/" + tool_name, params) def source_symbols(self, filenames): """Extract symbols (functions, classes, variables, types, etc.) from source code files using AI analysis. Accepts multiple files and returns structured symbol information for each. :param filenames: Array of file paths to analyze. """ params = {} params["filenames"] = filenames return self.execute_tool("source_symbols", **params) def xdg_open(self, path): """Open a file or directory using the system default application (e.g., PDF viewer, image viewer, file manager). :param path: Path to the file or directory to open. """ params = {} params["path"] = path return self.execute_tool("xdg_open", **params) def list_directory(self, path=None, pattern=None): """List contents of a directory with optional pattern filtering. :param path: The directory path to list (default: current directory). :param pattern: Filter pattern (default: *). """ params = {} if path is not None: params["path"] = path if pattern is not None: params["pattern"] = pattern return self.execute_tool("list_directory", **params) def fetch_url(self, url, strip_html=None): """Fetch the content of a URL as text. Useful for reading web pages, API responses, documentation, and other text content. HTML tags are stripped by default. :param url: The URL to fetch content from. :param strip_html: Strip HTML tags from response. Default: true. """ params = {} params["url"] = url if strip_html is not None: params["strip_html"] = strip_html return self.execute_tool("fetch_url", **params) def service_manage(self, action, auto_restart=None, auto_start=None, basic_auth_pass=None, basic_auth_user=None, command=None, config=None, environment=None, limit=None, local_port=None, name=None, ninja_user=None, restart_delay_ms=None, retry_limit=None, service_type=None, subdomain=None, working_directory=None): """Manage persistent services (tunnels and commands). Actions: create, start, stop, restart, delete, list, status, logs, update. :param action: Action to perform: create, start, stop, restart, delete, list, status, logs, update :param auto_restart: Auto-restart on crash :param auto_start: Auto-start on boot :param basic_auth_pass: Basic auth password for tunnel :param basic_auth_user: Basic auth username for tunnel :param command: Shell command for command services :param config: Full config JSON (alternative to individual fields) :param environment: Environment variables for command services :param limit: Log entry limit for logs action :param local_port: Local port for tunnel services :param name: Service name (required for all actions except list) :param ninja_user: Ninja user owning the service :param restart_delay_ms: Delay between retries in ms :param retry_limit: Max restart retries :param service_type: Service type: tunnel or command (required for create) :param subdomain: Subdomain for tunnel services :param working_directory: Working directory for command services """ params = {} params["action"] = action if auto_restart is not None: params["auto_restart"] = auto_restart if auto_start is not None: params["auto_start"] = auto_start if basic_auth_pass is not None: params["basic_auth_pass"] = basic_auth_pass if basic_auth_user is not None: params["basic_auth_user"] = basic_auth_user if command is not None: params["command"] = command if config is not None: params["config"] = config if environment is not None: params["environment"] = environment if limit is not None: params["limit"] = limit if local_port is not None: params["local_port"] = local_port if name is not None: params["name"] = name if ninja_user is not None: params["ninja_user"] = ninja_user if restart_delay_ms is not None: params["restart_delay_ms"] = restart_delay_ms if retry_limit is not None: params["retry_limit"] = retry_limit if service_type is not None: params["service_type"] = service_type if subdomain is not None: params["subdomain"] = subdomain if working_directory is not None: params["working_directory"] = working_directory return self.execute_tool("service_manage", **params) def cloud_share(self, path, expires_in=None, password=None, slug=None, username=None): """Share a local file or directory via Ninja Cloud with a public URL. Creates a static share accessible at static.ninja.molodetz.nl/{slug}. Only use when user explicitly mentions sharing via ninja cloud. :param path: Path to local file or directory to share :param expires_in: Expiration time in seconds from now (0 = never) :param password: Basic auth password for password protection :param slug: Custom URL slug (auto-generated if omitted) :param username: Basic auth username for password protection """ params = {} params["path"] = path if expires_in is not None: params["expires_in"] = expires_in if password is not None: params["password"] = password if slug is not None: params["slug"] = slug if username is not None: params["username"] = username return self.execute_tool("cloud_share", **params) def cloud_storage(self): """Show Ninja Cloud storage usage summary. Only use when user asks about their cloud storage. """ return self.execute_tool("cloud_storage") def websearch(self, query): """Search the web for current information. Use this for news, documentation, technology questions, or anything where up-to-date information matters. :param query: The search query string. """ params = {} params["query"] = query return self.execute_tool("websearch", **params) def download_file(self, path, url): """Download a file from a URL and save it to disk. Use this for binary files like images, archives, or any file that needs to be saved locally. :param path: The file path to save the downloaded file to. :param url: The URL to download from. """ params = {} params["path"] = path params["url"] = url return self.execute_tool("download_file", **params) def search_files(self, pattern, directory=None): """Search for files containing a specific pattern. :param pattern: The text pattern to search for. :param directory: The directory to search (default: current). """ params = {} params["pattern"] = pattern if directory is not None: params["directory"] = directory return self.execute_tool("search_files", **params) def execute_python(self, code): """Execute Python code and return the output. :param code: The Python code to execute. """ params = {} params["code"] = code return self.execute_tool("execute_python", **params) def prompt(self, task, context=None): """Send a task to an AI sub-conversation in a fresh context window. Useful for content processing, summarization, translation, or any task that benefits from a clean context. No tool access in sub-conversation. :param task: The task for the AI to complete. :param context: Additional context or data to include with the task. """ params = {} params["task"] = task if context is not None: params["context"] = context return self.execute_tool("prompt", **params) def upsert_tool(self, description, name, language=None, parameters=None, prompt=None, return_description=None, script_description=None, type=None): """Create or update a custom tool. Supports two types: 'script' (generates and executes Python/Bash scripts) and 'prompt' (AI-powered, sends system prompt to AI). Script tools are auto-generated, syntax-verified, and persisted. Prompt tools delegate to AI on invocation. :param description: What the tool does. Shown in tool listings. :param name: Unique tool name (lowercase, underscores). Must not conflict with built-in tools. :param language: Script language. Only used when type=script. Default: python. :param parameters: JSON schema for tool parameters. Optional. :param prompt: System prompt defining the tool's behavior. Required for type=prompt. :param return_description: Description of what the tool returns. Optional. :param script_description: Detailed description of what the script should do. Used for AI-powered script generation when type=script. Falls back to description if omitted. :param type: Tool type: 'script' for executable Python/Bash tools, 'prompt' for AI-powered tools. Default: prompt. """ params = {} params["description"] = description params["name"] = name if language is not None: params["language"] = language if parameters is not None: params["parameters"] = parameters if prompt is not None: params["prompt"] = prompt if return_description is not None: params["return_description"] = return_description if script_description is not None: params["script_description"] = script_description if type is not None: params["type"] = type return self.execute_tool("upsert_tool", **params) def generate_document(self, content, content_type, filename=None): """Generate a well-formatted document using AI. Supports markdown, html, json, yaml, xml, csv, and other text formats. Optionally saves to disk. :param content: Description of what the document should contain. :param content_type: Document format: markdown, html, json, yaml, xml, csv, etc. :param filename: File path to save the document. If omitted, content is returned directly. """ params = {} params["content"] = content params["content_type"] = content_type if filename is not None: params["filename"] = filename return self.execute_tool("generate_document", **params) def orchestrate(self, tasks, timeout_seconds=None): """Execute multiple tool calls in sequence with progress tracking and timeout. Use for batch operations that require running several tools. Each task specifies a tool name and its arguments. :param tasks: Array of {tool, args} objects to execute. :param timeout_seconds: Hard timeout for all tasks combined. Default: 300. """ params = {} params["tasks"] = tasks if timeout_seconds is not None: params["timeout_seconds"] = timeout_seconds return self.execute_tool("orchestrate", **params) def delete_file(self, path): """Delete a file at the given path. :param path: The file path to delete. """ params = {} params["path"] = path return self.execute_tool("delete_file", **params) def set_ninja_user(self, username): """Set the ninja user identity for this instance :param username: Username (alphanumeric, underscore, hyphen, max 64 chars) """ params = {} params["username"] = username return self.execute_tool("set_ninja_user", **params) def git_status(self, directory=None): """Show the working tree status (read-only git operation). :param directory: The git repository directory (default: current). """ params = {} if directory is not None: params["directory"] = directory return self.execute_tool("git_status", **params) def delete_tool(self, name): """Delete a custom tool. Asks for user confirmation before deleting. Removes both the JSON definition and any associated script file. :param name: Name of the custom tool to delete. """ params = {} params["name"] = name return self.execute_tool("delete_tool", **params) def port_info(self, port): """Get information about what process is using a given network port. :param port: The port number to check (1-65535). """ params = {} params["port"] = port return self.execute_tool("port_info", **params) def cloud_upload(self, path, name=None, type=None): """Upload a local directory or file to Ninja Cloud. Creates a versioned, git-backed cloud resource. Only use when user explicitly mentions 'ninja cloud' or cloud context. :param path: Path to local directory or file to upload :param name: Name for the cloud resource (defaults to directory name) :param type: Resource type: project, backup, or static (default: project) """ params = {} params["path"] = path if name is not None: params["name"] = name if type is not None: params["type"] = type return self.execute_tool("cloud_upload", **params) def describe_image(self, image, prompt): """Analyze an image and return structured JSON with description, objects, colors, text content, and confidence. Supports file paths and URLs. :param image: File path or URL of the image to analyze :param prompt: What to extract or describe from the image """ params = {} params["image"] = image params["prompt"] = prompt return self.execute_tool("describe_image", **params) def execute_plan(self, plan): """Execute a validated plan tree with dependency ordering, progress tracking, timeouts, and error isolation. Failed tasks do not crash the plan; dependents are skipped. :param plan: The JSON plan string from create_plan. """ params = {} params["plan"] = plan return self.execute_tool("execute_plan", **params) def ask_user(self, question, default_value=None, timeout_seconds=None): """Ask the user a question and wait for their response. Use only for critical clarifications where you cannot proceed without user input. Always provide a sensible default_value. :param question: The question to ask the user. :param default_value: Default answer if user presses Enter without typing. :param timeout_seconds: Timeout in seconds. Default: 60. """ params = {} params["question"] = question if default_value is not None: params["default_value"] = default_value if timeout_seconds is not None: params["timeout_seconds"] = timeout_seconds return self.execute_tool("ask_user", **params) def repair_file(self, content_type, filename): """Repair a file by fixing format errors using AI. Overwrites the file with corrected content and validates the result. :param content_type: Expected format: json, yaml, xml, html, markdown, csv, toml, ini, etc. :param filename: Path to the file to repair. """ params = {} params["content_type"] = content_type params["filename"] = filename return self.execute_tool("repair_file", **params) def create_plan(self, objective, context=None, execute=None): """Recursively decompose a complex objective into a validated, dependency-aware task tree with AI validation. Returns structured JSON plan. Use for deep, multi-step tasks with inter-task dependencies. :param objective: The complex objective to decompose into a task tree. :param context: Additional context, constraints, or requirements for planning. :param execute: If true, immediately execute the plan after creation. Default: false. """ params = {} params["objective"] = objective if context is not None: params["context"] = context if execute is not None: params["execute"] = execute return self.execute_tool("create_plan", **params) def git_log(self, directory=None, max_count=None): """Show commit logs (read-only git operation). :param directory: The git repository directory (default: current). :param max_count: Maximum number of commits to show (default: 10). """ params = {} if directory is not None: params["directory"] = directory if max_count is not None: params["max_count"] = max_count return self.execute_tool("git_log", **params) def plan(self, task_description): """Decompose a complex task into a structured sequence of subtasks using AI. Returns a JSON plan with tool calls, dependencies, and execution order. Use this before orchestrate for complex multi-step tasks. :param task_description: Detailed description of the task to decompose into subtasks. """ params = {} params["task_description"] = task_description return self.execute_tool("plan", **params) def validate_plan_results(self, plan, results): """Verify each task's output against its validation criteria using AI. Returns only failures with recommendations. :param plan: The original plan JSON string. :param results: The execution results JSON string from execute_plan. """ params = {} params["plan"] = plan params["results"] = results return self.execute_tool("validate_plan_results", **params) def cloud_list(self, type=None): """List cloud resources (projects, backups, static files) stored in Ninja Cloud. Only use when user asks about their cloud resources. :param type: Filter by type: project, backup, or static (all types if omitted) """ params = {} if type is not None: params["type"] = type return self.execute_tool("cloud_list", **params) def ask_user_form(self, fields, description=None, title=None): """Present a structured form to the user with multiple fields. Use for collecting multiple inputs at once. Field types: text, select, multiselect, confirm, number. :param fields: Array of field specs. Each field: {name, field_type, label, description, required, default, options (for select/multiselect), min/max/step (for number)}. :param description: Form description text. :param title: Form title. """ params = {} params["fields"] = fields if description is not None: params["description"] = description if title is not None: params["title"] = title return self.execute_tool("ask_user_form", **params) def folder_stats(self, path=None): """Analyze a directory and return comprehensive statistics: per-language lines of code, file categories (source, images, videos, executables, misc), project type detection, size breakdowns in human-readable and byte formats, modification dates, and path information. Skips dependency directories (node_modules, .venv, vendor, target, etc.) and only counts LOC for files under 200KB. :param path: The directory path to analyze (default: working directory). """ params = {} if path is not None: params["path"] = path return self.execute_tool("folder_stats", **params) def write_file(self, content, path): """Write content to a file at the given path, overwriting if exists. :param content: The content to write. :param path: The file path to write. """ params = {} params["content"] = content params["path"] = path return self.execute_tool("write_file", **params) def commit_message(self, directory=None): """Generate an AI-powered git commit message from the current diff in a directory. :param directory: Git repository directory (defaults to working directory) """ params = {} if directory is not None: params["directory"] = directory return self.execute_tool("commit_message", **params) def validate_file(self, content_type, filename): """Validate a file against its expected format using AI analysis. Returns validity status with errors and warnings. :param content_type: Expected format: json, yaml, xml, html, markdown, csv, toml, ini, etc. :param filename: Path to the file to validate. """ params = {} params["content_type"] = content_type params["filename"] = filename return self.execute_tool("validate_file", **params) def read_file(self, path): """Read the content of a file at the given path. :param path: The file path to read. """ params = {} params["path"] = path return self.execute_tool("read_file", **params) def port_kill(self, port): """Kill the process occupying a given network port. Sends SIGTERM first, then SIGKILL if needed. :param port: The port number whose process to kill (1-65535). """ params = {} params["port"] = port return self.execute_tool("port_kill", **params) def execute_shell(self, command): """Execute a shell command and return the output. :param command: The shell command to execute. """ params = {} params["command"] = command return self.execute_tool("execute_shell", **params) def git_diff(self, args=None, directory=None): """Show changes between commits (read-only git operation). :param args: Additional arguments to pass to git diff. :param directory: The git repository directory (default: current). """ params = {} if args is not None: params["args"] = args if directory is not None: params["directory"] = directory return self.execute_tool("git_diff", **params) def append_to_file(self, path, text): """Append text to a file at the given path. :param path: The file path to append to. :param text: The text to append. """ params = {} params["path"] = path params["text"] = text return self.execute_tool("append_to_file", **params) def create_directory(self, path, recursive=None): """Create a directory, optionally recursively creating parent directories. :param path: The directory path to create. :param recursive: Create parent directories (default: true). """ params = {} params["path"] = path if recursive is not None: params["recursive"] = recursive return self.execute_tool("create_directory", **params)