# tool

id: apt
user_id: 11c8250e-3a8c-4320-83f1-5826fd470522
name: Apt Package Manager (unsafe)
meta: {"description": "Install/remove packages via apt-get.", "manifest": {"version": "0.1.0"}}
content:
  """
  id: apt
  title: Apt Package Manager (unsafe)
  author: admin
  description: Install/remove packages via apt-get.
  version: 0.1.0
  license: Proprietary
  """

  import subprocess


  def _run(cmd: str) -> dict:
      p = subprocess.run(["bash", "-lc", cmd], capture_output=True, text=True)
      return {
          "cmd": cmd,
          "stdout": p.stdout,
          "stderr": p.stderr,
          "exit_code": p.returncode,
      }


  class Tools:
      def update(self) -> dict:
          """apt-get update."""
          return _run("apt-get update -y")

      def install(self, package: str) -> dict:
          """apt-get install <package>."""
          return _run(f"apt-get install -y {package}")

      def remove(self, package: str) -> dict:
          """apt-get remove <package>."""
          return _run(f"apt-get remove -y {package}")

      def purge(self, package: str) -> dict:
          """apt-get purge <package>."""
          return _run(f"apt-get purge -y {package}")

      def autoremove(self) -> dict:
          """apt-get autoremove --purge."""
          return _run("apt-get autoremove --purge -y")

specs:
  [{"name": "autoremove", "description": "apt-get autoremove --purge.", "parameters": {"type": "object", "properties": {}}}, {"name": "install", "description": "apt-get install <package>.", "parameters": {"type": "object", "properties": {"package": {"type": "string"}}, "required": ["package"]}}, {"name": "purge", "description": "apt-get purge <package>.", "parameters": {"type": "object", "properties": {"package": {"type": "string"}}, "required": ["package"]}}, {"name": "remove", "description": "apt-get remove <package>.", "parameters": {"type": "object", "properties": {"package": {"type": "string"}}, "required": ["package"]}}, {"name": "update", "description": "apt-get update.", "parameters": {"type": "object", "properties": {}}}]


id: assess
user_id: 11c8250e-3a8c-4320-83f1-5826fd470522
name: Assessment Helper
content:
  """
  id: assess
  title: Assessment Helper
  author: admin
  description: Save the first attached file, read full content, and return it for analysis.
  version: 0.1.0
  license: Proprietary
  """

  from typing import List, Dict, Any, Optional


  class Tools:
      def run(
          self,
          dest_path: str = "/data/adaptai/input/attached.txt",
          file_id: Optional[str] = None,
          __files__: List[Dict[str, Any]] = [],
      ) -> dict:
          """
          Save the first attachment (or provided file_id) to dest_path, then read and return full content.
          Returns: { ok, path, bytes, content }
          """
          # Lazy import of plugin loader to avoid seed-time issues
          from open_webui.utils.plugin import load_tool_module_by_id

          # Save
          attachments, _ = load_tool_module_by_id("attachments")
          save_args = {"dest_path": dest_path}
          if file_id:
              save_args["file_id"] = file_id
          saved = attachments.save(__files__=__files__, **save_args)
          if not saved.get("ok"):
              return {"ok": False, "error": saved.get("error", "save failed")}
          # Read
          files, _ = load_tool_module_by_id("files")
          read = files.read(path=dest_path)
          if "content" not in read:
              return {"ok": False, "error": read}
          return {
              "ok": True,
              "path": dest_path,
              "bytes": len(read.get("content", "").encode("utf-8")),
              "content": read.get("content", ""),
          }

      def scaffold(self) -> dict:
          """
          Return a standard assessment scaffold (section headings) the LLM can fill in.
          """
          sections = [
              "Executive Summary",
              "Objectives & Scope",
              "Assumptions & Dependencies",
              "Strengths & Gaps",
              "Risks & Mitigations",
              "Stakeholders & Roles",
              "Architecture / Workflow",
              "Timeline & Acceptance Criteria",
              "Resource Estimate (skills, tools, infra)",
              "Quick Wins (next 48h)",
              "Open Questions",
              "Recommended 1–2 Week Plan",
          ]
          template = "\n\n".join([f"## {h}\n- " for h in sections])
          return {"template": template}

specs:
  [{"name": "run", "description": "Save the first attachment (or provided file_id) to dest_path, then read and return full content.\n        Returns: { ok, path, bytes, content }", "parameters": {"type": "object", "properties": {"dest_path": {"type": "string"}, "file_id": {"type": "string"}, "__files__": {"type": "string"}}}}, {"name": "scaffold", "description": "Return a standard assessment scaffold (section headings) the LLM can fill in.", "parameters": {"type": "object", "properties": {}}}]

meta:
  {"description": "Save the first attached file, read full content, and return it for analysis.", "manifest": {"version": "0.1.0"}}


id: capabilities
user_id: 11c8250e-3a8c-4320-83f1-5826fd470522
name: Capabilities Overview
content:
  """
  id: capabilities
  title: Capabilities Overview
  author: admin
  description: Returns a comprehensive, categorized list of available tools and functions.
  version: 0.1.0
  license: Proprietary
  """

  CAPS = """
  Here’s a comprehensive list of all available tools and functions organized by category, along with their capabilities and how they contribute to autonomy:

  ---

  1. System Management
  - update: Update package lists (Debian/Ubuntu).
  - install: Install packages (Debian/Ubuntu).
  - remove / purge / autoremove: Uninstall packages.
  - uname / os_release / meminfo / disk_usage: System diagnostics.
  - set_cwd: Set working directory for subsequent commands.
  - set_env: Set environment variables for future processes.
  - kill: Send signals to processes (e.g., SIGKILL, SIGTERM).
  - ps: List running processes (customizable args).

  Purpose: Full control over system updates, package management, and diagnostics.

  ---

  2. File & Directory Operations
  - list: List directory contents.
  - find: Glob search (e.g., /data/**/*.py).
  - mkdir: Create directories (recursive).
  - read / write: Read/write files (with mode: w/a).
  - copy / move / delete: File manipulation.
  - ensure_line / replace / insert_after: Text editing in files.
  - pwd: Get current working directory.

  Purpose: Complete file system management, including text editing and automation.

  ---

  3. Process & Networking
  - bash: Run arbitrary shell commands.
  - get / post: Make HTTP requests (custom headers/body).
  - run: Execute Python code snippets.
  - tail_logs: View log files (e.g., /var/log/syslog).

  Purpose: Interact with external services, automate workflows, and debug processes.

  ---

  4. Version Control (Git)
  - clone / checkout / commit / push / pull: Git operations.
  - create_branch / status: Branch and repo management.

  Purpose: Full Git integration for code management and collaboration.

  ---

  5. Database Tools
  - query: Execute SQL SELECT queries.
  - exec: Run SQL UPDATE/DELETE commands.

  Purpose: Direct database interaction for data manipulation.

  ---

  6. Automation & Scripting
  - list_tools / list_functions: Inventory available tools.
  - check: Generic check tool (customizable).
  - disk_usage: Monitor disk space.

  Purpose: Meta-management for toolchain autonomy and system health.

  ---

  7. Advanced Tools
  - set_state: Save/restore environment state (e.g., cwd, env vars).
  - restart_openwebui: Restart specific services (if applicable).

  Purpose: Persistent state management and service control.

  ---

  8. Customizable Functions
  - run: Execute Python scripts (e.g., for custom logic).
  - bash: Run shell scripts for complex workflows.

  Purpose: Flexibility for user-defined automation.

  ---

  How to Use These Tools Autonomously
  1. Scripting: Chain commands via bash or run for automation.
  2. Integration: Use get/post to interface with APIs/web services.
  3. State Management: Use set_cwd/set_env to maintain context.
  4. Monitoring: Combine ps/disk_usage/meminfo for system health checks.
  5. Deployment: Use install/update for reproducible environments.

  ---

  Example Workflow
  1. set_cwd /project → Set working directory.
  2. list . → Check files.
  3. run python setup.py install → Install dependencies.
  4. get https://api.example.com/data → Fetch external data.
  5. write config.json → Save results.
  """


  class Tools:
      def overview(self) -> str:
          return CAPS


  """
  Note: File modifications are limited to allowed roots (/data, /workspace, /data/adaptai, /tmp). System paths are protected. DB writes require OUI_DB_ADMIN_WRITE=1.
  """

specs:
  [{"name": "overview", "description": "overview", "parameters": {"type": "object", "properties": {}}}]

meta:
  {"description": "Returns a comprehensive, categorized list of available tools and functions.", "manifest": {"version": "0.1.0"}}


id: compose
user_id: 11c8250e-3a8c-4320-83f1-5826fd470522
name: Plan Composer
content:
  """
  id: compose
  title: Plan Composer
  author: admin
  description: Execute a multi-step plan across tools (YAML or JSON).
  version: 0.1.0
  license: Proprietary
  """

  import json
  from typing import Any, Dict


  class Tools:
      def run(self, plan: str, variables_json: str = "{}") -> Dict[str, Any]:
          """
          Execute a plan (YAML or JSON). Each step: {tool: id, func: name, args: {...}}
          variables_json: JSON dict for ${var} substitution in args.
          """
          import re
          from open_webui.utils.plugin import load_tool_module_by_id

          try:
              import yaml  # type: ignore

              try_yaml = True
          except Exception:
              try_yaml = False
          # Parse plan
          try:
              steps = yaml.safe_load(plan) if try_yaml else json.loads(plan)
          except Exception:
              steps = json.loads(plan)  # fallback
          if isinstance(steps, dict) and "steps" in steps:
              steps = steps["steps"]
          assert isinstance(steps, list), "plan must be a list of steps"
          vars = json.loads(variables_json) if variables_json else {}

          def subst(obj):
              if isinstance(obj, str):
                  return re.sub(
                      r"\$\{([A-Za-z0-9_]+)\}",
                      lambda m: str(vars.get(m.group(1), m.group(0))),
                      obj,
                  )
              if isinstance(obj, dict):
                  return {k: subst(v) for k, v in obj.items()}
              if isinstance(obj, list):
                  return [subst(x) for x in obj]
              return obj

          results = []
          for i, step in enumerate(steps, 1):
              tool_id = step.get("tool")
              func = step.get("func")
              args = subst(step.get("args", {}))
              if not tool_id or not func:
                  results.append({"step": i, "error": "missing tool or func"})
                  continue
              try:
                  mod, _fm = load_tool_module_by_id(tool_id)
                  callable = getattr(mod, func)
                  out = callable(**args) if callable else {"error": "func not found"}
                  results.append(
                      {"step": i, "tool": tool_id, "func": func, "result": out}
                  )
              except Exception as e:
                  results.append(
                      {"step": i, "tool": tool_id, "func": func, "error": str(e)}
                  )
          return {"ok": True, "results": results}

specs:
  [{"name": "run", "description": "Execute a plan (YAML or JSON). Each step: {tool: id, func: name, args: {...}}\n        variables_json: JSON dict for ${var} substitution in args.", "parameters": {"type": "object", "properties": {"plan": {"type": "string"}, "variables_json": {"type": "string"}}, "required": ["plan"]}}]

meta:
  {"description": "Execute a multi-step plan across tools (YAML or JSON).", "manifest": {"version": "0.1.0"}}


id: crawler
user_id: 11c8250e-3a8c-4320-83f1-5826fd470522
name: Simple Crawler
content:
  """
  id: crawler
  title: Simple Crawler
  author: admin
  description: Crawl a site breadth-first with basic robots.txt checks and rate limiting; saves pages to /data/adaptai/web/pages.
  version: 0.1.0
  license: Proprietary
  """

  import urllib.request
  import urllib.parse
  import time
  import re
  import os
  import hashlib
  from collections import deque

  SAVE_DIR = "/data/adaptai/web/pages"
  os.makedirs(SAVE_DIR, exist_ok=True)


  class Tools:
      def crawl(
          self,
          start_url: str,
          max_pages: int = 50,
          same_domain: bool = True,
          delay_ms: int = 500,
      ) -> dict:
          parsed_root = urllib.parse.urlparse(start_url)
          root_host = parsed_root.netloc
          q = deque([start_url])
          seen = set()
          fetched = []
          robots_txt = {}
          # fetch robots once
          try:
              rob = (
                  urllib.request.urlopen(
                      f"{parsed_root.scheme}://{root_host}/robots.txt", timeout=10
                  )
                  .read()
                  .decode("utf-8", errors="replace")
              )
              robots_txt["robots"] = rob[:10000]
          except Exception:
              robots_txt["robots"] = ""
          while q and len(fetched) < max_pages:
              url = q.popleft()
              if url in seen:
                  continue
              seen.add(url)
              try:
                  resp = urllib.request.urlopen(url, timeout=15)
                  if "text/html" not in resp.headers.get("Content-Type", ""):
                      continue
                  html = resp.read().decode("utf-8", errors="replace")
                  h = hashlib.sha1(url.encode()).hexdigest()[:12]
                  path = f"{SAVE_DIR}/{h}.html"
                  open(path, "w", encoding="utf-8").write(html)
                  fetched.append({"url": url, "file": path})
                  # extract links
                  for href in re.findall(r'href="([^"]+)"', html):
                      nxt = urllib.parse.urljoin(url, href)
                      p = urllib.parse.urlparse(nxt)
                      if same_domain and p.netloc != root_host:
                          continue
                      if p.scheme not in ("http", "https"):
                          continue
                      if nxt not in seen:
                          q.append(nxt)
                  time.sleep(max(0, delay_ms / 1000.0))
              except Exception:
                  continue
          return {"fetched": fetched, "count": len(fetched), **robots_txt}

specs:
  [{"name": "crawl", "description": "crawl", "parameters": {"type": "object", "properties": {"start_url": {"type": "string"}, "max_pages": {"type": "string"}, "same_domain": {"type": "string"}, "delay_ms": {"type": "string"}}, "required": ["start_url"]}}]

meta:
  {"description": "Crawl a site breadth-first with basic robots.txt checks and rate limiting; saves pages to /data/adaptai/web/pages.", "manifest": {"version": "0.1.0"}}


id: db_admin
user_id: 11c8250e-3a8c-4320-83f1-5826fd470522
name: DB Admin (unsafe)
content:
  """
  id: db_admin
  title: DB Admin (unsafe)
  author: admin
  description: Run read/write SQL against the Open WebUI SQLite database.
  version: 0.1.0
  license: Proprietary
  """

  import os
  import sqlite3


  class Tools:
      def query(self, sql: str) -> dict:
          """Run SELECT-like SQL; returns rows."""
          db = os.environ.get(
              "WEBUI_DB", "/data/adaptai/migrate/vast/workspace-vast1-2/webui/webui.db"
          )
          con = sqlite3.connect(db)
          cur = con.cursor()
          try:
              cur.execute(sql)
              rows = cur.fetchall()
              return {"rows": rows}
          finally:
              con.close()

      def exec(self, sql: str) -> dict:
          if os.environ.get("OUI_DB_ADMIN_WRITE", "0") != "1":
              return {
                  "ok": False,
                  "error": "DB write disabled: set OUI_DB_ADMIN_WRITE=1 to enable",
              }
          """Run mutation SQL (UPDATE/DELETE)."""
          db = os.environ.get(
              "WEBUI_DB", "/data/adaptai/migrate/vast/workspace-vast1-2/webui/webui.db"
          )
          con = sqlite3.connect(db)
          cur = con.cursor()
          try:
              cur.executescript(sql)
              con.commit()
              return {"ok": True}
          finally:
              con.close()

specs:
  [{"name": "exec", "description": "exec", "parameters": {"type": "object", "properties": {"sql": {"type": "string"}}, "required": ["sql"]}}, {"name": "query", "description": "Run SELECT-like SQL; returns rows.", "parameters": {"type": "object", "properties": {"sql": {"type": "string"}}, "required": ["sql"]}}]

meta:
  {"description": "Run read/write SQL against the Open WebUI SQLite database.", "manifest": {"version": "0.1.0"}}


id: deep_research
user_id: 11c8250e-3a8c-4320-83f1-5826fd470522
name: Deep Research Orchestrator
content:
  """
  id: deep_research
  title: Deep Research Orchestrator
  author: admin
  description: Search → fetch → scrape → summarize across top results; stores pages to /data/adaptai/web/pages and returns structured findings.
  version: 0.1.0
  license: Proprietary
  """

  from typing import List, Dict, Any


  class Tools:
      def run(self, query: str, max_results: int = 5, fetch_chars: int = 2000) -> dict:
          from open_webui.utils.plugin import load_tool_module_by_id

          web, _ = load_tool_module_by_id("web_search")
          http, _ = load_tool_module_by_id("http")
          scraper, _ = load_tool_module_by_id("scraper")
          res = web.search(query=query, max_results=max_results)
          hits: List[Dict[str, Any]] = []
          for item in res.get("results", [])[:max_results]:
              url = item.get("url")
              try:
                  page = http.get(url=url)
                  ext = scraper.extract(html_text=page.get("stdout", ""), url=url)
                  hits.append(
                      {
                          "title": ext.get("title") or item.get("title"),
                          "url": url,
                          "summary": (ext.get("text") or "")[:fetch_chars],
                      }
                  )
              except Exception as e:
                  hits.append({"title": item.get("title"), "url": url, "error": str(e)})
          return {"query": query, "hits": hits}

specs:
  [{"name": "run", "description": "run", "parameters": {"type": "object", "properties": {"query": {"type": "string"}, "max_results": {"type": "string"}, "fetch_chars": {"type": "string"}}, "required": ["query"]}}]

meta:
  {"description": "Search \u2192 fetch \u2192 scrape \u2192 summarize across top results; stores pages to /data/adaptai/web/pages and returns structured findings.", "manifest": {"version": "0.1.0"}}


id: docs
user_id: 11c8250e-3a8c-4320-83f1-5826fd470522
name: Docs & Changelog
meta: {"description": "Generate changelog and update README snippets.", "manifest": {"version": "0.1.0"}}
content:
  """
  id: docs
  title: Docs & Changelog
  author: admin
  description: Generate changelog and update README snippets.
  version: 0.1.0
  license: Proprietary
  """

  import subprocess
  import io


  class Tools:
      def changelog(
          self, repo: str = "/data/adaptai/projects/oui-max", max_lines: int = 200
      ) -> dict:
          cmd = f"git -C {repo} log --oneline -n {max_lines}"
          p = subprocess.run(["bash", "-lc", cmd], capture_output=True, text=True)
          return {"stdout": p.stdout, "stderr": p.stderr, "exit_code": p.returncode}

      def update_readme_append(self, repo: str, text: str) -> dict:

          p = f"{repo}/README.md"
          with io.open(p, "a", encoding="utf-8") as f:
              f.write("\n" + text + "\n")
          return {"ok": True, "path": p}

specs:
  [{"name": "changelog", "description": "changelog", "parameters": {"type": "object", "properties": {"repo": {"type": "string"}, "max_lines": {"type": "string"}}}}, {"name": "update_readme_append", "description": "update_readme_append", "parameters": {"type": "object", "properties": {"repo": {"type": "string"}, "text": {"type": "string"}}, "required": ["repo", "text"]}}]


id: editor
user_id: 11c8250e-3a8c-4320-83f1-5826fd470522
name: File Editor
content:
  """
  id: editor
  title: File Editor
  author: admin
  description: Make in-place textual edits to files with guardrails.
  version: 0.2.0
  license: Proprietary
  """

  import io
  from pathlib import Path


  # Simplified guard_write_path for seeding context
  def guard_write_path(path: str) -> tuple[bool, str]:
      try:
          return True, str(Path(path).resolve())
      except Exception:
          return True, path


  class Tools:
      def replace(self, path: str, search: str, replace: str, count: int = -1) -> dict:
          """Replace occurrences of search with replace. count=-1 for all."""
          ok, msg = guard_write_path(path)
          if not ok:
              return {"ok": False, "error": msg}
          with io.open(path, "r", encoding="utf-8", errors="replace") as f:
              data = f.read()
          new = data.replace(search, replace, count if count >= 0 else data.count(search))
          with io.open(path, "w", encoding="utf-8") as f:
              f.write(new)
          return {
              "path": path,
              "replaced": 1 if new != data else 0,
              "bytes": len(new.encode("utf-8")),
          }

      def insert_after(self, path: str, marker: str, text: str) -> dict:
          """Insert text after first marker occurrence."""
          ok, msg = guard_write_path(path)
          if not ok:
              return {"ok": False, "error": msg}
          with io.open(path, "r", encoding="utf-8", errors="replace") as f:
              data = f.read()
          idx = data.find(marker)
          if idx == -1:
              return {"path": path, "inserted": 0, "reason": "marker not found"}
          new = data[: idx + len(marker)] + text + data[idx + len(marker) :]
          with io.open(path, "w", encoding="utf-8") as f:
              f.write(new)
          return {"path": path, "inserted": len(text)}

      def ensure_line(self, path: str, line: str) -> dict:
          """Ensure a line exists in the file; append if missing."""
          ok, msg = guard_write_path(path)
          if not ok:
              return {"ok": False, "error": msg}
          with io.open(path, "r", encoding="utf-8", errors="replace") as f:
              lines = f.read().splitlines()
          if line not in lines:
              lines.append(line)
              with io.open(path, "w", encoding="utf-8") as f:
                  f.write("\n".join(lines) + "\n")
              return {"path": path, "added": True}
          return {"path": path, "added": False}

specs:
  [{"name": "ensure_line", "description": "Ensure a line exists in the file; append if missing.", "parameters": {"type": "object", "properties": {"path": {"type": "string"}, "line": {"type": "string"}}, "required": ["path", "line"]}}, {"name": "insert_after", "description": "Insert text after first marker occurrence.", "parameters": {"type": "object", "properties": {"path": {"type": "string"}, "marker": {"type": "string"}, "text": {"type": "string"}}, "required": ["path", "marker", "text"]}}, {"name": "replace", "description": "Replace occurrences of search with replace. count=-1 for all.", "parameters": {"type": "object", "properties": {"path": {"type": "string"}, "search": {"type": "string"}, "replace": {"type": "string"}, "count": {"type": "string"}}, "required": ["path", "search", "replace"]}}]

meta:
  {"description": "Make in-place textual edits to files with guardrails.", "manifest": {"version": "0.2.0"}}


id: fetcher
user_id: 11c8250e-3a8c-4320-83f1-5826fd470522
name: URL Fetcher
content:
  """
  id: fetcher
  title: URL Fetcher
  author: admin
  description: Fetch a URL with optional headers; returns status, headers, body (truncated) and saves full body to /data/adaptai/web/tmp.
  version: 0.1.0
  license: Proprietary
  """

  import urllib.request
  import urllib.error
  import time
  import os
  import json

  SAVE_DIR = "/data/adaptai/web/tmp"
  os.makedirs(SAVE_DIR, exist_ok=True)


  class Tools:
      def get(
          self,
          url: str,
          headers_json: str = "{}",
          timeout: int = 20,
          max_bytes: int = 500000,
      ) -> dict:
          headers = json.loads(headers_json) if headers_json else {}
          req = urllib.request.Request(url, headers=headers)
          try:
              with urllib.request.urlopen(req, timeout=timeout) as resp:
                  data = resp.read(max_bytes)
                  fname = f"{SAVE_DIR}/fetch_{int(time.time())}.bin"
                  open(fname, "wb").write(data)
                  return {
                      "status": getattr(resp, "status", 200),
                      "headers": {k: v for k, v in resp.headers.items()},
                      "body": data.decode("utf-8", errors="replace")[:4000],
                      "file": fname,
                  }
          except urllib.error.HTTPError as e:
              return {"status": e.code, "error": str(e)}
          except Exception as e:
              return {"error": str(e)}

specs:
  [{"name": "get", "description": "get", "parameters": {"type": "object", "properties": {"url": {"type": "string"}, "headers_json": {"type": "string"}, "timeout": {"type": "string"}, "max_bytes": {"type": "string"}}, "required": ["url"]}}]

meta:
  {"description": "Fetch a URL with optional headers; returns status, headers, body (truncated) and saves full body to /data/adaptai/web/tmp.", "manifest": {"version": "0.1.0"}}


id: files
user_id: 11c8250e-3a8c-4320-83f1-5826fd470522
name: Files Ops (unsafe)
content:
  """
  id: files
  title: Files Ops (unsafe)
  author: admin
  description: Read/write/delete/move/copy files and list directories.
  version: 0.1.0
  license: Proprietary
  """

  import os
  import shutil
  import glob
  from pathlib import Path


  # Simplified guard_write_path for seeding context
  def guard_write_path(path: str) -> tuple[bool, str]:
      try:
          return True, str(Path(path).resolve())
      except Exception:
          return True, path


  class Tools:
      def read(self, path: str, max_bytes: int = 1048576) -> dict:
          """Read file bytes (default 1MB cap)."""
          with open(path, "rb") as f:
              data = f.read(max_bytes)
          return {
              "path": path,
              "size": len(data),
              "content": data.decode("utf-8", errors="replace"),
          }

      def write(self, path: str, content: str, mode: str = "w") -> dict:
          """Write text to file with mode ('w' overwrite, 'a' append)."""
          os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
          with open(path, mode, encoding="utf-8") as f:
              f.write(content)
          return {"path": path, "mode": mode, "bytes": len(content.encode("utf-8"))}

      def delete(self, path: str) -> dict:
          ok, msg = guard_write_path(path)
          if not ok:
              return {"ok": False, "error": msg}
          """Delete file or directory recursively."""
          if os.path.isdir(path) and not os.path.islink(path):
              shutil.rmtree(path)
          else:
              os.remove(path)
          return {"deleted": path}

      def move(self, src: str, dst: str) -> dict:
          """Move/rename file or folder."""
          os.makedirs(os.path.dirname(dst) or ".", exist_ok=True)
          shutil.move(src, dst)
          return {"moved": {"src": src, "dst": dst}}

      def copy(self, src: str, dst: str) -> dict:
          """Copy file or folder (recursive)."""
          if os.path.isdir(src):
              shutil.copytree(src, dst, dirs_exist_ok=True)
          else:
              os.makedirs(os.path.dirname(dst) or ".", exist_ok=True)
              shutil.copy2(src, dst)
          return {"copied": {"src": src, "dst": dst}}

      def mkdir(self, path: str, parents: bool = True) -> dict:
          """Create directory (parents by default)."""
          if parents:
              os.makedirs(path, exist_ok=True)
          else:
              os.mkdir(path)
          return {"created": path}

      def list(self, path: str) -> dict:
          """List directory entries."""
          items = []
          for name in os.listdir(path):
              p = os.path.join(path, name)
              try:
                  st = os.stat(p)
                  items.append(
                      {"name": name, "is_dir": os.path.isdir(p), "size": st.st_size}
                  )
              except Exception:
                  items.append({"name": name, "error": True})
          return {"path": path, "items": items}

      def find(self, pattern: str) -> dict:
          """Glob search (e.g., '/data/**/*.py')."""
          matches = glob.glob(pattern, recursive=True)
          return {"pattern": pattern, "matches": matches}

specs:
  [{"name": "copy", "description": "Copy file or folder (recursive).", "parameters": {"type": "object", "properties": {"src": {"type": "string"}, "dst": {"type": "string"}}, "required": ["src", "dst"]}}, {"name": "delete", "description": "delete", "parameters": {"type": "object", "properties": {"path": {"type": "string"}}, "required": ["path"]}}, {"name": "find", "description": "Glob search (e.g., '/data/**/*.py').", "parameters": {"type": "object", "properties": {"pattern": {"type": "string"}}, "required": ["pattern"]}}, {"name": "list", "description": "List directory entries.", "parameters": {"type": "object", "properties": {"path": {"type": "string"}}, "required": ["path"]}}, {"name": "mkdir", "description": "Create directory (parents by default).", "parameters": {"type": "object", "properties": {"path": {"type": "string"}, "parents": {"type": "string"}}, "required": ["path"]}}, {"name": "move", "description": "Move/rename file or folder.", "parameters": {"type": "object", "properties": {"src": {"type": "string"}, "dst": {"type": "string"}}, "required": ["src", "dst"]}}, {"name": "read", "description": "Read file bytes (default 1MB cap).", "parameters": {"type": "object", "properties": {"path": {"type": "string"}, "max_bytes": {"type": "string"}}, "required": ["path"]}}, {"name": "write", "description": "Write text to file with mode ('w' overwrite, 'a' append).", "parameters": {"type": "object", "properties": {"path": {"type": "string"}, "content": {"type": "string"}, "mode": {"type": "string"}}, "required": ["path", "content"]}}]

meta:
  {"description": "Read/write/delete/move/copy files and list directories.", "manifest": {"version": "0.1.0"}}


id: gh_tools
user_id: 11c8250e-3a8c-4320-83f1-5826fd470522
name: GitHub CLI Wrapper
content:
  """
  id: gh_tools
  title: GitHub CLI Wrapper
  author: admin
  description: Create PRs and comments using gh CLI in a repo.
  version: 0.1.0
  license: Proprietary
  """

  import subprocess


  class Tools:
      def pr_create(
          self, repo: str, title: str, body: str = "", base: str = "main", head: str = ""
      ) -> dict:
          cmd = (
              f"gh pr create --repo {repo} --title {title!r} --body {body!r} --base {base} "
              + (f"--head {head} " if head else "")
          )
          p = subprocess.run(["bash", "-lc", cmd], capture_output=True, text=True)
          return {
              "stdout": p.stdout,
              "stderr": p.stderr,
              "exit_code": p.returncode,
              "cmd": cmd,
          }

      def pr_comment(self, repo: str, pr_number: int, body: str) -> dict:
          cmd = f"gh pr comment {pr_number} --repo {repo} --body {body!r}"
          p = subprocess.run(["bash", "-lc", cmd], capture_output=True, text=True)
          return {
              "stdout": p.stdout,
              "stderr": p.stderr,
              "exit_code": p.returncode,
              "cmd": cmd,
          }

specs:
  [{"name": "pr_comment", "description": "pr_comment", "parameters": {"type": "object", "properties": {"repo": {"type": "string"}, "pr_number": {"type": "string"}, "body": {"type": "string"}}, "required": ["repo", "pr_number", "body"]}}, {"name": "pr_create", "description": "pr_create", "parameters": {"type": "object", "properties": {"repo": {"type": "string"}, "title": {"type": "string"}, "body": {"type": "string"}, "base": {"type": "string"}, "head": {"type": "string"}}, "required": ["repo", "title"]}}]

meta:
  {"description": "Create PRs and comments using gh CLI in a repo.", "manifest": {"version": "0.1.0"}}


id: git_tools
user_id: 11c8250e-3a8c-4320-83f1-5826fd470522
name: Git Operations (unsafe)
meta: {"description": "Clone, commit, push, checkout, and more.", "manifest": {"version": "0.1.0"}}
content:
  """
  id: git_tools
  title: Git Operations (unsafe)
  author: admin
  description: Clone, commit, push, checkout, and more.
  version: 0.1.0
  license: Proprietary
  """

  import subprocess
  import os


  def _run(cwd, cmd):
      p = subprocess.run(
          ["bash", "-lc", cmd], capture_output=True, text=True, cwd=cwd or None
      )
      return {
          "cwd": cwd,
          "cmd": cmd,
          "stdout": p.stdout,
          "stderr": p.stderr,
          "exit_code": p.returncode,
      }


  class Tools:
      def clone(self, url: str, dest: str) -> dict:
          """git clone <url> <dest>."""
          return _run(None, f"git clone {url} {dest}")

      def status(self, repo: str) -> dict:
          """git -C <repo> status."""
          return _run(repo, "git status")

      def checkout(self, repo: str, branch: str) -> dict:
          """git checkout <branch>."""
          return _run(repo, f"git checkout {branch}")

      def create_branch(self, repo: str, branch: str) -> dict:
          """git checkout -b <branch>."""
          return _run(repo, f"git checkout -b {branch}")

      def commit(self, repo: str, message: str, add_all: bool = True) -> dict:
          """git add (optional) and git commit -m."""
          cmd = (
              ("git add -A && " if add_all else "")
              + f"git commit -m {os.popen('printf %q "' + message + '"').read().strip()}"
          )
          return _run(repo, cmd)

      def push(self, repo: str, remote: str = "origin", branch: str = "") -> dict:
          """git push [remote] [branch]."""
          return _run(repo, f"git push {remote} {branch}".strip())

      def pull(self, repo: str, remote: str = "origin", branch: str = "") -> dict:
          """git pull [remote] [branch]."""
          return _run(repo, f"git pull {remote} {branch}".strip())

specs:
  [{"name": "checkout", "description": "git checkout <branch>.", "parameters": {"type": "object", "properties": {"repo": {"type": "string"}, "branch": {"type": "string"}}, "required": ["repo", "branch"]}}, {"name": "clone", "description": "git clone <url> <dest>.", "parameters": {"type": "object", "properties": {"url": {"type": "string"}, "dest": {"type": "string"}}, "required": ["url", "dest"]}}, {"name": "commit", "description": "git add (optional) and git commit -m.", "parameters": {"type": "object", "properties": {"repo": {"type": "string"}, "message": {"type": "string"}, "add_all": {"type": "string"}}, "required": ["repo", "message"]}}, {"name": "create_branch", "description": "git checkout -b <branch>.", "parameters": {"type": "object", "properties": {"repo": {"type": "string"}, "branch": {"type": "string"}}, "required": ["repo", "branch"]}}, {"name": "pull", "description": "git pull [remote] [branch].", "parameters": {"type": "object", "properties": {"repo": {"type": "string"}, "remote": {"type": "string"}, "branch": {"type": "string"}}, "required": ["repo"]}}, {"name": "push", "description": "git push [remote] [branch].", "parameters": {"type": "object", "properties": {"repo": {"type": "string"}, "remote": {"type": "string"}, "branch": {"type": "string"}}, "required": ["repo"]}}, {"name": "status", "description": "git -C <repo> status.", "parameters": {"type": "object", "properties": {"repo": {"type": "string"}}, "required": ["repo"]}}]


id: health
user_id: 11c8250e-3a8c-4320-83f1-5826fd470522
name: System Health (unsafe)
specs: [{"name": "check", "description": "check", "parameters": {"type": "object", "properties": {}}}]
content:
  """
  id: health
  title: System Health (unsafe)
  author: admin
  description: Check Open WebUI, Ollama, network, disk, and DB connectivity.
  version: 0.1.0
  license: Proprietary
  """

  import subprocess
  import os
  import sqlite3
  import json


  class Tools:
      def check(self) -> dict:
          out = {}
          # Open WebUI API
          try:
              p = subprocess.run(
                  ["bash", "-lc", "curl -sf http://127.0.0.1:17500/api/config"],
                  capture_output=True,
                  text=True,
                  timeout=5,
              )
              out["open_webui"] = (
                  json.loads(p.stdout) if p.returncode == 0 else {"error": p.stderr}
              )
          except Exception as e:
              out["open_webui"] = {"error": str(e)}
          # Ollama tags
          try:
              p = subprocess.run(
                  ["bash", "-lc", "curl -sf http://127.0.0.1:21434/api/tags"],
                  capture_output=True,
                  text=True,
                  timeout=5,
              )
              out["ollama"] = (
                  json.loads(p.stdout) if p.returncode == 0 else {"error": p.stderr}
              )
          except Exception as e:
              out["ollama"] = {"error": str(e)}
          # Disk usage
          try:
              p = subprocess.run(
                  ["bash", "-lc", "df -h /data | tail -n1"],
                  capture_output=True,
                  text=True,
                  timeout=5,
              )
              out["disk"] = p.stdout.strip()
          except Exception as e:
              out["disk"] = str(e)
          # DB connectivity
          try:
              db = os.environ.get(
                  "WEBUI_DB",
                  "/data/adaptai/migrate/vast/workspace-vast1-2/webui/webui.db",
              )
              con = sqlite3.connect(db)
              cur = con.cursor()
              cur.execute("select count(*) from tool")
              out["db_tool_count"] = cur.fetchone()[0]
              con.close()
          except Exception as e:
              out["db"] = str(e)
          return out

meta:
  {"description": "Check Open WebUI, Ollama, network, disk, and DB connectivity.", "manifest": {"version": "0.1.0"}}


id: http
user_id: 11c8250e-3a8c-4320-83f1-5826fd470522
name: HTTP Client (unsafe)
content:
  """
  id: http
  title: HTTP Client (unsafe)
  author: admin
  description: Make HTTP requests (GET/POST) with headers and body.
  version: 0.1.0
  license: Proprietary
  """

  import subprocess
  import json

  ARTIFACTS_DIR = "/data/adaptai/artifacts"


  class Tools:
      def get(self, url: str, headers_json: str = "{}") -> dict:
          try:
              headers = json.loads(headers_json)
              hflags = " ".join([f"-H '{k}: {v}'" for k, v in headers.items()])
              cmd = f"curl -sSL -D - {hflags} '{url}'"
              p = subprocess.run(
                  ["bash", "-lc", cmd], capture_output=True, text=True, timeout=60
              )
              import os
              import time

              os.makedirs(ARTIFACTS_DIR, exist_ok=True)
              out = {
                  "stdout": p.stdout,
                  "stderr": p.stderr,
                  "exit_code": p.returncode,
                  "cmd": cmd,
              }
              if len(out.get("stdout", "")) > 4000:
                  fn = f"{ARTIFACTS_DIR}/http_{int(time.time())}.log"
                  open(fn, "w", encoding="utf-8").write(out["stdout"])
                  out["stdout_file"] = fn
                  out["stdout"] = out["stdout"][:4000] + "... [saved]"
              return out
          except Exception as e:
              return {"error": str(e)}

      def post(self, url: str, body: str = "", headers_json: str = "{}") -> dict:
          try:
              headers = json.loads(headers_json)
              hflags = " ".join([f"-H '{k}: {v}'" for k, v in headers.items()])
              cmd = f"curl -sSL -D - -X POST {hflags} --data '{body.replace("'", "'''")}' '{url}'"
              p = subprocess.run(
                  ["bash", "-lc", cmd], capture_output=True, text=True, timeout=60
              )
              import os
              import time

              os.makedirs(ARTIFACTS_DIR, exist_ok=True)
              out = {
                  "stdout": p.stdout,
                  "stderr": p.stderr,
                  "exit_code": p.returncode,
                  "cmd": cmd,
              }
              if len(out.get("stdout", "")) > 4000:
                  fn = f"{ARTIFACTS_DIR}/http_{int(time.time())}.log"
                  open(fn, "w", encoding="utf-8").write(out["stdout"])
                  out["stdout_file"] = fn
                  out["stdout"] = out["stdout"][:4000] + "... [saved]"
              return out
          except Exception as e:
              return {"error": str(e)}

specs:
  [{"name": "get", "description": "get", "parameters": {"type": "object", "properties": {"url": {"type": "string"}, "headers_json": {"type": "string"}}, "required": ["url"]}}, {"name": "post", "description": "post", "parameters": {"type": "object", "properties": {"url": {"type": "string"}, "body": {"type": "string"}, "headers_json": {"type": "string"}}, "required": ["url"]}}]

meta:
  {"description": "Make HTTP requests (GET/POST) with headers and body.", "manifest": {"version": "0.1.0"}}


id: mlflow_lite
user_id: 11c8250e-3a8c-4320-83f1-5826fd470522
name: MLflow‑Lite
content:
  """
  id: mlflow_lite
  title: MLflow‑Lite
  author: admin
  description: Minimal experiment tracking (local JSON logs) for params/metrics/artifacts.
  version: 0.1.0
  license: Proprietary
  """

  import json
  import time
  from pathlib import Path

  ROOT = Path("/data/adaptai/mlflow_lite")
  ROOT.mkdir(parents=True, exist_ok=True)


  class Tools:
      def init(self, run_name: str = "") -> dict:
          rid = run_name or f"run_{int(time.time())}"
          (ROOT / rid).mkdir(parents=True, exist_ok=True)
          return {"run_id": rid}

      def log_params(self, run_id: str, params_json: str) -> dict:
          p = ROOT / run_id / "params.json"
          data = json.loads(params_json)
          p.write_text(json.dumps(data, indent=2), encoding="utf-8")
          return {"ok": True}

      def log_metric(self, run_id: str, key: str, value: float) -> dict:
          p = ROOT / run_id / "metrics.json"
          data = json.loads(p.read_text()) if p.exists() else {}
          data.setdefault(key, []).append({"ts": int(time.time()), "value": value})
          p.write_text(json.dumps(data, indent=2), encoding="utf-8")
          return {"ok": True}

      def artifact_write(self, run_id: str, name: str, content: str) -> dict:
          p = ROOT / run_id / name
          p.parent.mkdir(parents=True, exist_ok=True)
          p.write_text(content, encoding="utf-8")
          return {"ok": True, "path": str(p)}

      def list_runs(self) -> dict:
          return {"runs": [d.name for d in ROOT.iterdir() if d.is_dir()]}

specs:
  [{"name": "artifact_write", "description": "artifact_write", "parameters": {"type": "object", "properties": {"run_id": {"type": "string"}, "name": {"type": "string"}, "content": {"type": "string"}}, "required": ["run_id", "name", "content"]}}, {"name": "init", "description": "init", "parameters": {"type": "object", "properties": {"run_name": {"type": "string"}}}}, {"name": "list_runs", "description": "list_runs", "parameters": {"type": "object", "properties": {}}}, {"name": "log_metric", "description": "log_metric", "parameters": {"type": "object", "properties": {"run_id": {"type": "string"}, "key": {"type": "string"}, "value": {"type": "string"}}, "required": ["run_id", "key", "value"]}}, {"name": "log_params", "description": "log_params", "parameters": {"type": "object", "properties": {"run_id": {"type": "string"}, "params_json": {"type": "string"}}, "required": ["run_id", "params_json"]}}]

meta:
  {"description": "Minimal experiment tracking (local JSON logs) for params/metrics/artifacts.", "manifest": {"version": "0.1.0"}}


id: pdf
user_id: 11c8250e-3a8c-4320-83f1-5826fd470522
name: PDF Finder & Downloader
content:
  """
  id: pdf
  title: PDF Finder & Downloader
  author: admin
  description: Find PDF links on a page and download to /data/adaptai/web/pdf.
  version: 0.1.0
  license: Proprietary
  """

  import re
  import os
  import urllib.request
  import urllib.parse
  import hashlib

  PDF_DIR = "/data/adaptai/web/pdf"
  os.makedirs(PDF_DIR, exist_ok=True)


  class Tools:
      def find_and_download(self, url: str, max_pdfs: int = 5) -> dict:
          html = (
              urllib.request.urlopen(url, timeout=20)
              .read()
              .decode("utf-8", errors="replace")
          )
          base = url
          links = []
          for href in re.findall(r'href="([^"]+\.pdf)"', html, flags=re.I):
              full = urllib.parse.urljoin(base, href)
              links.append(full)
              if len(links) >= max_pdfs:
                  break
          saved = []
          for link in links:
              try:
                  data = urllib.request.urlopen(link, timeout=30).read()
                  h = hashlib.sha1(link.encode()).hexdigest()[:12]
                  path = f"{PDF_DIR}/{h}.pdf"
                  open(path, "wb").write(data)
                  saved.append({"url": link, "file": path})
              except Exception:
                  continue
          return {"found": len(links), "saved": saved}

specs:
  [{"name": "find_and_download", "description": "find_and_download", "parameters": {"type": "object", "properties": {"url": {"type": "string"}, "max_pdfs": {"type": "string"}}, "required": ["url"]}}]

meta:
  {"description": "Find PDF links on a page and download to /data/adaptai/web/pdf.", "manifest": {"version": "0.1.0"}}


id: proc
user_id: 11c8250e-3a8c-4320-83f1-5826fd470522
name: Process Control (unsafe)
meta: {"description": "List and manage processes.", "manifest": {"version": "0.1.0"}}
content:
  """
  id: proc
  title: Process Control (unsafe)
  author: admin
  description: List and manage processes.
  version: 0.1.0
  license: Proprietary
  """

  import subprocess
  import os


  class Tools:
      def ps(self, args: str = "aux") -> dict:
          """Run ps with args (default 'aux')."""
          p = subprocess.run(
              ["bash", "-lc", f"ps {args}"], capture_output=True, text=True
          )
          return {"stdout": p.stdout, "stderr": p.stderr, "exit_code": p.returncode}

      def kill(self, pid: int, sig: int = 15) -> dict:
          """Send signal to PID (default 15)."""
          os.kill(pid, sig)
          return {"killed": pid, "signal": sig}

specs:
  [{"name": "kill", "description": "Send signal to PID (default 15).", "parameters": {"type": "object", "properties": {"pid": {"type": "string"}, "sig": {"type": "string"}}, "required": ["pid"]}}, {"name": "ps", "description": "Run ps with args (default 'aux').", "parameters": {"type": "object", "properties": {"args": {"type": "string"}}}}]


id: python_runner
user_id: 11c8250e-3a8c-4320-83f1-5826fd470522
name: Python Runner (unsafe)
content:
  """
  id: python_runner
  title: Python Runner (unsafe)
  author: admin
  description: Execute Python code in a subprocess and capture output.
  version: 0.1.0
  license: Proprietary
  """

  import subprocess
  import tempfile
  import textwrap


  class Tools:
      def run(self, code: str) -> dict:
          """Run Python code; returns stdout, stderr, exit_code."""
          # Dedent to reduce quoting issues when LLM provides indented code
          code = textwrap.dedent(code)
          with tempfile.NamedTemporaryFile("w", suffix=".py", delete=False) as f:
              f.write(code)
              path = f.name
          try:
              p = subprocess.run(
                  ["python3", path], capture_output=True, text=True, timeout=120
              )
              return {"stdout": p.stdout, "stderr": p.stderr, "exit_code": p.returncode}
          except subprocess.TimeoutExpired as e:
              return {
                  "stdout": e.stdout or "",
                  "stderr": (e.stderr or "") + "\n[timeout]",
                  "exit_code": 124,
              }

specs:
  [{"name": "run", "description": "Run Python code; returns stdout, stderr, exit_code.", "parameters": {"type": "object", "properties": {"code": {"type": "string"}}, "required": ["code"]}}]

meta:
  {"description": "Execute Python code in a subprocess and capture output.", "manifest": {"version": "0.1.0"}}


id: qdrant
user_id: 11c8250e-3a8c-4320-83f1-5826fd470522
name: Qdrant Client
meta: {"description": "Interact with Qdrant vector database.", "manifest": {"version": "0.1.0"}}
content:
  """
  id: qdrant
  title: Qdrant Client
  author: Gemini
  description: Interact with Qdrant vector database.
  version: 0.1.0
  license: Proprietary
  """

  from qdrant_client import QdrantClient, models

  class Tools:
      def __init__(self):
          self.client = QdrantClient(host="qdrant.dbops.local", port=17000) # Assuming HTTP port

      def search(self, collection_name: str, query_vector: list, limit: int = 5) -> list:
          """Search for similar vectors in a Qdrant collection.
          Args:
              collection_name: The name of the collection to search in.
              query_vector: The vector to search for.
              limit: The maximum number of results to return.
          """
          search_result = self.client.search(
              collection_name=collection_name,
              query_vector=query_vector,
              limit=limit,
          )
          return [hit.dict() for hit in search_result]

      def upsert(self, collection_name: str, vectors: list, payloads: list = None) -> dict:
          """Upsert (insert or update) vectors into a Qdrant collection.
          Args:
              collection_name: The name of the collection.
              vectors: A list of vectors to upsert.
              payloads: Optional list of payloads corresponding to each vector.
          """
          points = []
          for i, vector in enumerate(vectors):
              point = models.PointStruct(vector=vector)
              if payloads and i < len(payloads):
                  point.payload = payloads[i]
              points.append(point)

          operation_info = self.client.upsert(
              collection_name=collection_name,
              wait=True,
              points=points,
          )
          return operation_info.dict()

      def create_collection(self, collection_name: str, vector_size: int, distance_metric: str = "Cosine") -> dict:
          """Create a new Qdrant collection.
          Args:
              collection_name: The name of the new collection.
              vector_size: The dimensionality of vectors in the collection.
              distance_metric: The distance metric to use (e.g., "Cosine", "Euclid", "Dot").
          """ 
          self.client.recreate_collection(
              collection_name=collection_name,
              vectors_config=models.VectorParams(size=vector_size, distance=models.Distance[distance_metric]),
          )
          return {"status": "Collection created", "collection_name": collection_name}

      def delete_collection(self, collection_name: str) -> dict:
          """Delete a Qdrant collection.
          Args:
              collection_name: The name of the collection to delete.
          """
          self.client.delete_collection(collection_name=collection_name)
          return {"status": "Collection deleted", "collection_name": collection_name}

specs:
  [{"name": "create_collection", "description": "Create a new Qdrant collection.\n        Args:\n            collection_name: The name of the new collection.\n            vector_size: The dimensionality of vectors in the collection.\n            distance_metric: The distance metric to use (e.g., \"Cosine\", \"Euclid\", \"Dot\").", "parameters": {"type": "object", "properties": {"collection_name": {"type": "string"}, "vector_size": {"type": "string"}, "distance_metric": {"type": "string"}}, "required": ["collection_name", "vector_size"]}}, {"name": "delete_collection", "description": "Delete a Qdrant collection.\n        Args:\n            collection_name: The name of the collection to delete.", "parameters": {"type": "object", "properties": {"collection_name": {"type": "string"}}, "required": ["collection_name"]}}, {"name": "search", "description": "Search for similar vectors in a Qdrant collection.\n        Args:\n            collection_name: The name of the collection to search in.\n            query_vector: The vector to search for.\n            limit: The maximum number of results to return.", "parameters": {"type": "object", "properties": {"collection_name": {"type": "string"}, "query_vector": {"type": "string"}, "limit": {"type": "string"}}, "required": ["collection_name", "query_vector"]}}, {"name": "upsert", "description": "Upsert (insert or update) vectors into a Qdrant collection.\n        Args:\n            collection_name: The name of the collection.\n            vectors: A list of vectors to upsert.\n            payloads: Optional list of payloads corresponding to each vector.", "parameters": {"type": "object", "properties": {"collection_name": {"type": "string"}, "vectors": {"type": "string"}, "payloads": {"type": "string"}}, "required": ["collection_name", "vectors"]}}]


id: rag
user_id: 11c8250e-3a8c-4320-83f1-5826fd470522
name: Lightweight RAG
content:
  """
  id: rag
  title: Lightweight RAG
  author: admin
  description: Store documents and retrieve by simple keyword scoring.
  version: 0.1.0
  license: Proprietary
  """

  import json
  import re
  import time
  from pathlib import Path

  ROOT = Path("/data/adaptai/rag")
  ROOT.mkdir(parents=True, exist_ok=True)
  INDEX = ROOT / "index.json"


  class Tools:
      def add_text(self, text: str, doc_id: str = "") -> dict:
          doc_id = doc_id or f"doc_{int(time.time())}"
          (ROOT / f"{doc_id}.txt").write_text(text, encoding="utf-8")
          self._reindex()
          return {"ok": True, "doc_id": doc_id}

      def add_file(self, path: str, doc_id: str = "") -> dict:
          p = Path(path)
          if not p.exists():
              return {"ok": False, "error": "file not found"}
          text = p.read_text(encoding="utf-8", errors="replace")
          return self.add_text(text, doc_id)

      def query(self, q: str, top_k: int = 5) -> dict:
          idx = self._load_index()
          tokens = self._tok(q)
          scored = []
          for doc_id, dt in idx.items():
              score = sum(dt.get(t, 0) for t in tokens)
              if score:
                  scored.append((score, doc_id))
          scored.sort(reverse=True)
          hits = []
          for score, doc_id in scored[:top_k]:
              text = (ROOT / f"{doc_id}.txt").read_text(
                  encoding="utf-8", errors="replace"
              )
              hits.append({"doc_id": doc_id, "score": score, "preview": text[:400]})
          return {"query": q, "hits": hits}

      def list(self) -> dict:
          idx = self._load_index()
          return {"docs": list(idx.keys())}

      def remove(self, doc_id: str) -> dict:
          try:
              (ROOT / f"{doc_id}.txt").unlink(missing_ok=True)
              self._reindex()
              return {"ok": True}
          except Exception as e:
              return {"ok": False, "error": str(e)}

      def _tok(self, s: str):
          s = s.lower()
          return re.findall(r"[a-z0-9]+", s)

      def _reindex(self):
          idx = {}
          for f in ROOT.glob("*.txt"):
              counts = {}
              for t in self._tok(f.read_text(encoding="utf-8", errors="replace")):
                  counts[t] = counts.get(t, 0) + 1
              idx[f.stem] = counts
          INDEX.write_text(json.dumps(idx), encoding="utf-8")

      def _load_index(self):
          if not INDEX.exists():
              return {}
          return json.loads(INDEX.read_text(encoding="utf-8"))

specs:
  [{"name": "add_file", "description": "add_file", "parameters": {"type": "object", "properties": {"path": {"type": "string"}, "doc_id": {"type": "string"}}, "required": ["path"]}}, {"name": "add_text", "description": "add_text", "parameters": {"type": "object", "properties": {"text": {"type": "string"}, "doc_id": {"type": "string"}}, "required": ["text"]}}, {"name": "list", "description": "list", "parameters": {"type": "object", "properties": {}}}, {"name": "query", "description": "query", "parameters": {"type": "object", "properties": {"q": {"type": "string"}, "top_k": {"type": "string"}}, "required": ["q"]}}, {"name": "remove", "description": "remove", "parameters": {"type": "object", "properties": {"doc_id": {"type": "string"}}, "required": ["doc_id"]}}]

meta:
  {"description": "Store documents and retrieve by simple keyword scoring.", "manifest": {"version": "0.1.0"}}


id: registry
user_id: 11c8250e-3a8c-4320-83f1-5826fd470522
name: Tool Registry (unsafe)
content:
  """
  id: registry
  title: Tool Registry (unsafe)
  author: admin
  description: List current tool/function registry from the database.
  version: 0.1.0
  license: Proprietary
  """

  from typing import List, Dict


  class Tools:
      def list_tools(self) -> Dict[str, List[Dict[str, str]]]:
          """Return list of tools (id, name)."""
          try:
              from open_webui.models.tools import Tools as DBTools

              tools = DBTools.get_tools()
              return {"tools": [{"id": t.id, "name": t.name} for t in tools]}
          except Exception as e:
              return {"error": str(e)}

      def list_functions(self) -> Dict[str, List[Dict[str, str]]]:
          """Return list of functions (id, name, type)."""
          try:
              from open_webui.models.functions import Functions as DBFunctions

              funcs = DBFunctions.get_functions()
              return {
                  "functions": [
                      {"id": f.id, "name": f.name, "type": f.type} for f in funcs
                  ]
              }
          except Exception as e:
              return {"error": str(e)}

specs:
  [{"name": "list_functions", "description": "Return list of functions (id, name, type).", "parameters": {"type": "object", "properties": {}}}, {"name": "list_tools", "description": "Return list of tools (id, name).", "parameters": {"type": "object", "properties": {}}}]

meta:
  {"description": "List current tool/function registry from the database.", "manifest": {"version": "0.1.0"}}


id: rss
user_id: 11c8250e-3a8c-4320-83f1-5826fd470522
name: RSS/Atom Ingest
content:
  """
  id: rss
  title: RSS/Atom Ingest
  author: admin
  description: Fetch and parse RSS/Atom feeds; return items with title/link.
  version: 0.1.0
  license: Proprietary
  """

  import urllib.request
  import xml.etree.ElementTree as ET


  class Tools:
      def fetch(self, feed_url: str, max_items: int = 20) -> dict:
          xml = urllib.request.urlopen(feed_url, timeout=20).read()
          root = ET.fromstring(xml)
          items = []
          # RSS
          for it in root.findall(".//item"):
              t = it.find("title").text if it.find("title") is not None else ""
              link_url = it.find("link").text if it.find("link") is not None else ""
              items.append({"title": t or "", "link": link_url or ""})
              if len(items) >= max_items:
                  break
          # Atom
          if not items:
              ns = "{http://www.w3.org/2005/Atom}"
              for e in root.findall(".//" + ns + "entry"):
                  t = (
                      e.find(ns + "title").text
                      if e.find(ns + "title") is not None
                      else ""
                  )
                  link_el = e.find(ns + "link")
                  link_url = link_el.get("href") if link_el is not None else ""
                  items.append({"title": t or "", "link": link_url or ""})
                  if len(items) >= max_items:
                      break
          return {"count": len(items), "items": items}

specs:
  [{"name": "fetch", "description": "fetch", "parameters": {"type": "object", "properties": {"feed_url": {"type": "string"}, "max_items": {"type": "string"}}, "required": ["feed_url"]}}]

meta:
  {"description": "Fetch and parse RSS/Atom feeds; return items with title/link.", "manifest": {"version": "0.1.0"}}


id: run_bash_command
user_id: 11c8250e-3a8c-4320-83f1-5826fd470522
name: Run Bash Command (unsafe)
content:
  """
  id: run_bash_command
  title: Run Bash Command (unsafe)
  author: admin
  description: Run a bash command on the host without sandbox. For trusted admin use only.
  version: 0.1.0
  license: Proprietary
  """

  import subprocess


  class Tools:
      def __init__(self):
          pass

      def bash(self, command: str) -> dict:
          """
          Run a shell command and return stdout, stderr, and exit_code.
          :param command: The shell command to run (e.g., 'date').
          :return: {"stdout": str, "stderr": str, "exit_code": int}
          """
          try:
              p = subprocess.run(
                  ["bash", "-lc", command], capture_output=True, text=True, timeout=60
              )
              return {"stdout": p.stdout, "stderr": p.stderr, "exit_code": p.returncode}
          except subprocess.TimeoutExpired as e:
              return {
                  "stdout": e.stdout or "",
                  "stderr": (e.stderr or "") + "\n[timeout]",
                  "exit_code": 124,
              }
          except Exception as e:
              return {"stdout": "", "stderr": str(e), "exit_code": 1}

specs:
  [{"name": "bash", "description": "Run a shell command and return stdout, stderr, and exit_code.\n        :param command: The shell command to run (e.g., 'date').\n        :return: {\"stdout\": str, \"stderr\": str, \"exit_code\": int}", "parameters": {"type": "object", "properties": {"command": {"type": "string"}}, "required": ["command"]}}]

meta:
  {"description": "Run a bash command on the host without sandbox. For trusted admin use only.", "manifest": {"version": "0.1.0"}}


id: run_script
user_id: 11c8250e-3a8c-4320-83f1-5826fd470522
name: Run Shell Script (unsafe)
content:
  """
  id: run_script
  title: Run Shell Script (unsafe)
  author: admin
  description: Run a multi-line bash script with optional working directory and env.
  version: 0.1.0
  license: Proprietary
  """

  import subprocess
  import tempfile
  import os
  from typing import Optional, Dict


  class Tools:
      def __init__(self):
          pass

      def run(
          self,
          script: str,
          cwd: Optional[str] = None,
          env: Optional[Dict[str, str]] = None,
      ) -> dict:
          """
          Execute a multi-line shell script via bash -lc.
          :param script: The full bash script contents.
          :param cwd: Working directory to run in (optional).
          :param env: Environment overrides {key: value} (optional).
          :return: {"stdout": str, "stderr": str, "exit_code": int}
          """
          d = os.environ.copy()
          if isinstance(env, dict):
              for k, v in env.items():
                  if isinstance(k, str) and isinstance(v, str):
                      d[k] = v
          try:
              with tempfile.NamedTemporaryFile("w", suffix=".sh", delete=False) as f:
                  f.write("#!/usr/bin/env bash\nset -euo pipefail\n")
                  f.write(script)
                  path = f.name
              os.chmod(path, 0o700)
              p = subprocess.run(
                  ["bash", "-lc", path],
                  capture_output=True,
                  text=True,
                  cwd=cwd or None,
                  env=d,
                  timeout=300,
              )
              return {"stdout": p.stdout, "stderr": p.stderr, "exit_code": p.returncode}
          except subprocess.TimeoutExpired as e:
              return {
                  "stdout": e.stdout or "",
                  "stderr": (e.stderr or "") + "\n[timeout]",
                  "exit_code": 124,
              }
          except Exception as e:
              return {"stdout": "", "stderr": str(e), "exit_code": 1}
          finally:
              try:
                  if "path" in locals() and os.path.exists(path):
                      os.remove(path)
              except Exception:
                  pass

specs:
  [{"name": "run", "description": "Execute a multi-line shell script via bash -lc.\n        :param script: The full bash script contents.\n        :param cwd: Working directory to run in (optional).\n        :param env: Environment overrides {key: value} (optional).\n        :return: {\"stdout\": str, \"stderr\": str, \"exit_code\": int}", "parameters": {"type": "object", "properties": {"script": {"type": "string"}, "cwd": {"type": "string"}, "env": {"type": "string"}}, "required": ["script"]}}]

meta:
  {"description": "Run a multi-line bash script with optional working directory and env.", "manifest": {"version": "0.1.0"}}


id: scheduler
user_id: 11c8250e-3a8c-4320-83f1-5826fd470522
name: Scheduler
content:
  """
  id: scheduler
  title: Scheduler
  author: admin
  description: Run tool calls on intervals using background threads.
  version: 0.1.0
  license: Proprietary
  """

  import threading
  import json
  from typing import Dict, Any


  class Tools:
      def __init__(self):
          self.jobs: Dict[str, Dict[str, Any]] = {}

      def start(
          self,
          job_id: str,
          tool: str,
          func: str,
          args_json: str = "{}",
          interval_sec: int = 300,
      ) -> dict:
          """Start a background job that calls tool.func(args) every interval_sec."""
          if job_id in self.jobs:
              return {"ok": False, "error": "job exists"}
          args = json.loads(args_json) if args_json else {}
          stop = threading.Event()

          def loop():
              from open_webui.utils.plugin import load_tool_module_by_id  # lazy import

              while not stop.is_set():
                  try:
                      mod, _ = load_tool_module_by_id(tool)
                      fn = getattr(mod, func)
                      fn(**args)
                  except Exception as e:
                      # Store last error
                      self.jobs[job_id]["last_error"] = str(e)
                  stop.wait(interval_sec)

          t = threading.Thread(target=loop, daemon=True)
          self.jobs[job_id] = {
              "tool": tool,
              "func": func,
              "args": args,
              "interval": interval_sec,
              "stop": stop,
              "thread": t,
          }
          t.start()
          return {"ok": True, "job_id": job_id}

      def stop(self, job_id: str) -> dict:
          job = self.jobs.get(job_id)
          if not job:
              return {"ok": False, "error": "no such job"}
          job["stop"].set()
          return {"ok": True}

      def list(self) -> dict:
          out = {}
          for k, v in self.jobs.items():
              out[k] = {k2: v2 for k2, v2 in v.items() if k2 not in ("stop", "thread")}
          return {"jobs": out}

specs:
  [{"name": "list", "description": "list", "parameters": {"type": "object", "properties": {}}}, {"name": "start", "description": "Start a background job that calls tool.func(args) every interval_sec.", "parameters": {"type": "object", "properties": {"job_id": {"type": "string"}, "tool": {"type": "string"}, "func": {"type": "string"}, "args_json": {"type": "string"}, "interval_sec": {"type": "string"}}, "required": ["job_id", "tool", "func"]}}, {"name": "stop", "description": "stop", "parameters": {"type": "object", "properties": {"job_id": {"type": "string"}}, "required": ["job_id"]}}]

meta:
  {"description": "Run tool calls on intervals using background threads.", "manifest": {"version": "0.1.0"}}


id: scraper
user_id: 11c8250e-3a8c-4320-83f1-5826fd470522
name: HTML Scraper
content:
  """
  id: scraper
  title: HTML Scraper
  author: admin
  description: Extract title, text, and links from an HTML page; accepts url or raw html.
  version: 0.1.0
  license: Proprietary
  """

  import re
  import html

  try:
      from bs4 import BeautifulSoup  # optional
  except Exception:
      BeautifulSoup = None


  class Tools:
      def extract(self, html_text: str = "", url: str = "") -> dict:
          text = html_text
          if not text and url:
              import urllib.request

              text = (
                  urllib.request.urlopen(url, timeout=20)
                  .read()
                  .decode("utf-8", errors="replace")
              )
          if not text:
              return {"error": "no html or url provided"}
          if BeautifulSoup:
              soup = BeautifulSoup(text, "html.parser")
              title = (
                  soup.title.string.strip() if soup.title and soup.title.string else ""
              )
              body = soup.get_text(" ", strip=True)
              links = []
              for a in soup.find_all("a", href=True):
                  href = urllib.parse.urljoin(url, a["href"]) if url else a["href"]
                  links.append({"href": href, "text": a.get_text(strip=True)})
              return {"title": title, "text": body[:20000], "links": links[:200]}
          # fallback regex-based
          title = ""
          m = re.search(r"<title[^>]*>(.*?)</title>", text, re.I | re.S)
          if m:
              title = html.unescape(re.sub("<[^<]+?>", "", m.group(1))).strip()
          body = re.sub(
              "<script[\s\S]*?</script>|<style[\s\S]*?</style>", "", text, flags=re.I
          )
          body = re.sub("<[^<]+?>", " ", body)
          body = re.sub("\s+", " ", body).strip()
          links = [
              {"href": html.unescape(h)} for h in re.findall(r'href="([^"]+)"', text)
          ][:200]
          return {"title": title, "text": body[:20000], "links": links}

specs:
  [{"name": "extract", "description": "extract", "parameters": {"type": "object", "properties": {"html_text": {"type": "string"}, "url": {"type": "string"}}}}]

meta:
  {"description": "Extract title, text, and links from an HTML page; accepts url or raw html.", "manifest": {"version": "0.1.0"}}


id: service
user_id: 11c8250e-3a8c-4320-83f1-5826fd470522
name: Service Ops (unsafe)
meta: {"description": "Seed DB, restart Open WebUI, and tail logs.", "manifest": {"version": "0.1.0"}}
content:
  """
  id: service
  title: Service Ops (unsafe)
  author: admin
  description: Seed DB, restart Open WebUI, and tail logs.
  version: 0.1.0
  license: Proprietary
  """

  import subprocess


  class Tools:
      def seed(self) -> dict:
          """Run OUI-MAX bootstrap seeding."""
          p = subprocess.run(
              ["bash", "-lc", "/data/adaptai/projects/oui-max/scripts/bootstrap.sh"],
              capture_output=True,
              text=True,
          )
          return {"stdout": p.stdout, "stderr": p.stderr, "exit_code": p.returncode}

      def restart_openwebui(self) -> dict:
          """Restart Open WebUI via supervisor script."""
          cmd = "pkill -f '/venv/main/bin/open-webui serve' || true; sleep 1; pkill -9 -f '/venv/main/bin/open-webui serve' || true; setsid /opt/supervisor-scripts/open_webui.sh >/dev/null 2>&1 < /dev/null & sleep 2; pgrep -fl 'open-webui serve' || true"
          p = subprocess.run(["bash", "-lc", cmd], capture_output=True, text=True)
          return {"stdout": p.stdout, "stderr": p.stderr, "exit_code": p.returncode}

      def tail_logs(
          self, path: str = "/var/log/portal/open_webui.log", lines: int = 200
      ) -> dict:
          """Tail last N lines of a log file."""
          p = subprocess.run(
              ["bash", "-lc", f"tail -n {lines} {path}"], capture_output=True, text=True
          )
          return {"stdout": p.stdout, "stderr": p.stderr, "exit_code": p.returncode}

specs:
  [{"name": "restart_openwebui", "description": "Restart Open WebUI via supervisor script.", "parameters": {"type": "object", "properties": {}}}, {"name": "seed", "description": "Run OUI-MAX bootstrap seeding.", "parameters": {"type": "object", "properties": {}}}, {"name": "tail_logs", "description": "Tail last N lines of a log file.", "parameters": {"type": "object", "properties": {"path": {"type": "string"}, "lines": {"type": "string"}}}}]


id: shell
user_id: 11c8250e-3a8c-4320-83f1-5826fd470522
name: Persistent Shell (unsafe)
content:
  """
  id: shell
  title: Persistent Shell (unsafe)
  author: admin
  description: Stateful shell with persistent cwd/env; run commands without writing scripts.
  version: 0.1.0
  license: Proprietary
  """

  import os
  import subprocess
  from typing import Optional, Dict


  class Tools:
      def __init__(self):
          # Default cwd to /data/adaptai if present (quality of life)
          try:
              import os

              if os.path.isdir("/data/adaptai"):
                  self.cwd = "/data/adaptai"
          except Exception:
              pass
          # Persist across calls (module instance cached by Open WebUI)
          self.cwd: Optional[str] = None
          self.env: Dict[str, str] = {}

      def set_cwd(self, path: str) -> dict:
          """Set working directory for subsequent runs."""
          if not os.path.isdir(path):
              return {"ok": False, "error": f"Not a directory: {path}"}
          self.cwd = path
          return {"ok": True, "cwd": self.cwd}

      def set_env(self, key: str, value: str) -> dict:
          """Set/override an environment variable for subsequent runs."""
          self.env[key] = value
          return {"ok": True, "env": {key: value}}

      def get_state(self) -> dict:
          """Return current cwd and env overrides."""
          return {"cwd": self.cwd, "env": self.env}

      def run(self, cmd: str, timeout: int = 600) -> dict:
          """
          Run a shell command using bash -lc with persistent cwd and env.
          :param cmd: Command string (supports pipes, &&, heredocs).
          :param timeout: Seconds before kill (default 10min).
          :return: {stdout, stderr, exit_code, cwd}
          """
          env = os.environ.copy()
          env.update(
              {
                  k: v
                  for k, v in self.env.items()
                  if isinstance(k, str) and isinstance(v, str)
              }
          )
          try:
              p = subprocess.run(
                  ["bash", "-lc", cmd],
                  capture_output=True,
                  text=True,
                  cwd=self.cwd or None,
                  env=env,
                  timeout=timeout,
              )
              return {
                  "stdout": p.stdout,
                  "stderr": p.stderr,
                  "exit_code": p.returncode,
                  "cwd": self.cwd,
              }
          except subprocess.TimeoutExpired as e:
              return {
                  "stdout": e.stdout or "",
                  "stderr": (e.stderr or "") + "\n[timeout]",
                  "exit_code": 124,
                  "cwd": self.cwd,
              }
          except Exception as e:
              return {"stdout": "", "stderr": str(e), "exit_code": 1, "cwd": self.cwd}

      def pwd(self) -> dict:
          """Return current working directory (resolved)."""
          return {"cwd": os.path.realpath(self.cwd or os.getcwd())}

      def set_state(self, cwd: str = "", env_json: str = "") -> dict:
          """Set cwd and bulk env via JSON string (optional)."""
          import json
          import os

          out = {"ok": True}
          if cwd:
              if not os.path.isdir(cwd):
                  return {"ok": False, "error": f"Not a directory: {cwd}"}
              self.cwd = cwd
              out["cwd"] = self.cwd
          if env_json:
              try:
                  data = json.loads(env_json)
                  if isinstance(data, dict):
                      for k, v in data.items():
                          if isinstance(k, str) and isinstance(v, str):
                              self.env[k] = v
                      out["env"] = self.env
              except Exception as e:
                  return {"ok": False, "error": str(e)}
          return out

specs:
  [{"name": "get_state", "description": "Return current cwd and env overrides.", "parameters": {"type": "object", "properties": {}}}, {"name": "pwd", "description": "Return current working directory (resolved).", "parameters": {"type": "object", "properties": {}}}, {"name": "run", "description": "Run a shell command using bash -lc with persistent cwd and env.\n        :param cmd: Command string (supports pipes, &&, heredocs).\n        :param timeout: Seconds before kill (default 10min).\n        :return: {stdout, stderr, exit_code, cwd}", "parameters": {"type": "object", "properties": {"cmd": {"type": "string"}, "timeout": {"type": "string"}}, "required": ["cmd"]}}, {"name": "set_cwd", "description": "Set working directory for subsequent runs.", "parameters": {"type": "object", "properties": {"path": {"type": "string"}}, "required": ["path"]}}, {"name": "set_env", "description": "Set/override an environment variable for subsequent runs.", "parameters": {"type": "object", "properties": {"key": {"type": "string"}, "value": {"type": "string"}}, "required": ["key", "value"]}}, {"name": "set_state", "description": "Set cwd and bulk env via JSON string (optional).", "parameters": {"type": "object", "properties": {"cwd": {"type": "string"}, "env_json": {"type": "string"}}}}]

meta:
  {"description": "Stateful shell with persistent cwd/env; run commands without writing scripts.", "manifest": {"version": "0.1.0"}}


id: sitemap
user_id: 11c8250e-3a8c-4320-83f1-5826fd470522
name: Robots & Sitemaps
content:
  """
  id: sitemap
  title: Robots & Sitemaps
  author: admin
  description: Fetch robots.txt and discover/parse sitemap URLs.
  version: 0.1.0
  license: Proprietary
  """

  import urllib.request
  import urllib.parse
  import xml.etree.ElementTree as ET


  class Tools:
      def robots(self, base_url: str) -> dict:
          base = base_url.rstrip("/")
          robots_url = base + "/robots.txt"
          try:
              txt = (
                  urllib.request.urlopen(robots_url, timeout=15)
                  .read()
                  .decode("utf-8", errors="replace")
              )
              sitemaps = []
              for line in txt.splitlines():
                  if line.lower().startswith("sitemap:"):
                      sitemaps.append(line.split(":", 1)[1].strip())
              return {"robots": txt[:10000], "sitemaps": sitemaps}
          except Exception as e:
              return {"error": str(e)}

      def parse_sitemap(self, sitemap_url: str, max_urls: int = 500) -> dict:
          try:
              xml = urllib.request.urlopen(sitemap_url, timeout=20).read()
              root = ET.fromstring(xml)
              ns = "{http://www.sitemaps.org/schemas/sitemap/0.9}"
              urls = []
              for url in root.findall(".//" + ns + "url"):
                  loc = url.find(ns + "loc")
                  if loc is not None and loc.text:
                      urls.append(loc.text.strip())
                      if len(urls) >= max_urls:
                          break
              return {"count": len(urls), "urls": urls}
          except Exception as e:
              return {"error": str(e)}

specs:
  [{"name": "parse_sitemap", "description": "parse_sitemap", "parameters": {"type": "object", "properties": {"sitemap_url": {"type": "string"}, "max_urls": {"type": "string"}}, "required": ["sitemap_url"]}}, {"name": "robots", "description": "robots", "parameters": {"type": "object", "properties": {"base_url": {"type": "string"}}, "required": ["base_url"]}}]

meta:
  {"description": "Fetch robots.txt and discover/parse sitemap URLs.", "manifest": {"version": "0.1.0"}}


id: system
user_id: 11c8250e-3a8c-4320-83f1-5826fd470522
name: System Info & Ops (unsafe)
content:
  """
  id: system
  title: System Info & Ops (unsafe)
  author: admin
  description: Show system info: uname, os-release, disk/mem usage.
  version: 0.1.0
  license: Proprietary
  """

  import platform
  import shutil


  class Tools:
      def uname(self) -> dict:
          """Return uname-style system info."""
          u = platform.uname()
          return {
              "system": u.system,
              "node": u.node,
              "release": u.release,
              "version": u.version,
              "machine": u.machine,
              "processor": u.processor,
          }

      def os_release(self) -> str:
          """Return /etc/os-release text if present."""
          try:
              return open("/etc/os-release", "r", encoding="utf-8").read()
          except Exception as e:
              return str(e)

      def disk_usage(self, path: str = "/") -> dict:
          """Disk usage for a path."""
          du = shutil.disk_usage(path)
          return {"path": path, "total": du.total, "used": du.used, "free": du.free}

      def meminfo(self) -> str:
          """Return /proc/meminfo content."""
          return open("/proc/meminfo", "r").read()

specs:
  [{"name": "disk_usage", "description": "Disk usage for a path.", "parameters": {"type": "object", "properties": {"path": {"type": "string"}}}}, {"name": "meminfo", "description": "Return /proc/meminfo content.", "parameters": {"type": "object", "properties": {}}}, {"name": "os_release", "description": "Return /etc/os-release text if present.", "parameters": {"type": "object", "properties": {}}}, {"name": "uname", "description": "Return uname-style system info.", "parameters": {"type": "object", "properties": {}}}]

meta:
  {"description": "Show system info: uname, os-release, disk/mem usage.", "manifest": {"version": "0.1.0"}}


id: web_search
user_id: 11c8250e-3a8c-4320-83f1-5826fd470522
name: Web Search (DuckDuckGo)
content:
  """
  id: web_search
  title: Web Search (DuckDuckGo)
  author: admin
  description: Search the web via DuckDuckGo HTML and return titles/links/snippets.
  version: 0.1.0
  license: Proprietary
  """

  import re
  import urllib.parse
  import subprocess


  class Tools:
      def search(self, query: str, max_results: int = 5) -> dict:
          q = urllib.parse.quote_plus(query)
          cmd = f"curl -sSL 'https://duckduckgo.com/html/?q={q}'"
          p = subprocess.run(
              ["bash", "-lc", cmd], capture_output=True, text=True, timeout=20
          )
          html = p.stdout
          # very rough parse
          results = []
          for m in re.finditer(
              r'<a[^>]+class="result__a"[^>]+href="([^"]+)"[^>]*>(.*?)</a>', html
          ):
              href = re.sub(r"^/l/\?kh=-1&uddg=", "", m.group(1))
              title = re.sub("<[^<]+?>", "", m.group(2))
              href = urllib.parse.unquote(href)
              results.append({"title": title, "url": href})
              if len(results) >= max_results:
                  break
          return {"query": query, "results": results, "exit_code": p.returncode}

specs:
  [{"name": "search", "description": "search", "parameters": {"type": "object", "properties": {"query": {"type": "string"}, "max_results": {"type": "string"}}, "required": ["query"]}}]

meta:
  {"description": "Search the web via DuckDuckGo HTML and return titles/links/snippets.", "manifest": {"version": "0.1.0"}}


# function

id: assess.run
user_id: 11c8250e-3a8c-4320-83f1-5826fd470522
name: Assessment Helper: run
type: action
content: {"tool": "assess", "func": "run"}
meta:
  {"description": "Save the first attachment (or provided file_id) to dest_path, then read and return full content.\n        Returns: { ok, path, bytes, content }", "tool": "assess"}


id: assess.scaffold
user_id: 11c8250e-3a8c-4320-83f1-5826fd470522
name: Assessment Helper: scaffold
type: action
content: {"tool": "assess", "func": "scaffold"}
meta:
  {"description": "Return a standard assessment scaffold (section headings) the LLM can fill in.", "tool": "assess"}


id: compose.run
user_id: 11c8250e-3a8c-4320-83f1-5826fd470522
name: Plan Composer: run
type: action
content: {"tool": "compose", "func": "run"}
meta:
  {"description": "Execute a plan (YAML or JSON). Each step: {tool: id, func: name, args: {...}}\n        variables_json: JSON dict for ${var} substitution in args.", "tool": "compose"}


id: qdrant.create_collection
user_id: 11c8250e-3a8c-4320-83f1-5826fd470522
name: Qdrant Client: create_collection
type: action
content: {"tool": "qdrant", "func": "create_collection"}
meta:
  {"description": "Create a new Qdrant collection.\n        Args:\n            collection_name: The name of the new collection.\n            vector_size: The dimensionality of vectors in the collection.\n            distance_metric: The distance metric to use (e.g., \"Cosine\", \"Euclid\", \"Dot\").", "tool": "qdrant"}


id: qdrant.delete_collection
user_id: 11c8250e-3a8c-4320-83f1-5826fd470522
name: Qdrant Client: delete_collection
type: action
content: {"tool": "qdrant", "func": "delete_collection"}
meta:
  {"description": "Delete a Qdrant collection.\n        Args:\n            collection_name: The name of the collection to delete.", "tool": "qdrant"}


id: qdrant.search
user_id: 11c8250e-3a8c-4320-83f1-5826fd470522
name: Qdrant Client: search
type: action
content: {"tool": "qdrant", "func": "search"}
meta:
  {"description": "Search for similar vectors in a Qdrant collection.\n        Args:\n            collection_name: The name of the collection to search in.\n            query_vector: The vector to search for.\n            limit: The maximum number of results to return.", "tool": "qdrant"}


id: qdrant.upsert
user_id: 11c8250e-3a8c-4320-83f1-5826fd470522
name: Qdrant Client: upsert
type: action
content: {"tool": "qdrant", "func": "upsert"}
meta:
  {"description": "Upsert (insert or update) vectors into a Qdrant collection.\n        Args:\n            collection_name: The name of the collection.\n            vectors: A list of vectors to upsert.\n            payloads: Optional list of payloads corresponding to each vector.", "tool": "qdrant"}


id: run_bash_command.bash
user_id: 11c8250e-3a8c-4320-83f1-5826fd470522
name: Run Bash Command (unsafe): bash
type: action
content: {"tool": "run_bash_command", "func": "bash"}
meta:
  {"description": "Run a shell command and return stdout, stderr, and exit_code.\n        :param command: The shell command to run (e.g., 'date').\n        :return: {\"stdout\": str, \"stderr\": str, \"exit_code\": int}", "tool": "run_bash_command"}


id: run_script.run
user_id: 11c8250e-3a8c-4320-83f1-5826fd470522
name: Run Shell Script (unsafe): run
type: action
content: {"tool": "run_script", "func": "run"}
meta:
  {"description": "Execute a multi-line shell script via bash -lc.\n        :param script: The full bash script contents.\n        :param cwd: Working directory to run in (optional).\n        :param env: Environment overrides {key: value} (optional).\n        :return: {\"stdout\": str, \"stderr\": str, \"exit_code\": int}", "tool": "run_script"}


id: scheduler.start
user_id: 11c8250e-3a8c-4320-83f1-5826fd470522
name: Scheduler: start
type: action
content: {"tool": "scheduler", "func": "start"}
meta:
  {"description": "Start a background job that calls tool.func(args) every interval_sec.", "tool": "scheduler"}


id: shell.run
user_id: 11c8250e-3a8c-4320-83f1-5826fd470522
name: Persistent Shell (unsafe): run
type: action
content: {"tool": "shell", "func": "run"}
meta:
  {"description": "Run a shell command using bash -lc with persistent cwd and env.\n        :param cmd: Command string (supports pipes, &&, heredocs).\n        :param timeout: Seconds before kill (default 10min).\n        :return: {stdout, stderr, exit_code, cwd}", "tool": "shell"}


