openweb_ui_push_tool.py

#!/usr/bin/env python3
"""
Auther: Christopher Gray
Version: 0.1.6
Updated: 5/16/2026

Updated from: https://raw.githubusercontent.com/c2theg/ai/refs/heads/main/openweb_ui_push_tool.py

Push openwebui_tool.py to a running Open WebUI instance via its REST API.
No browser required — run this after every edit.

Auth (use one — email/password is easiest if you can't find your API key):
    export OWUI_URL="http://your-server:3000"

    Option A — email + password (works with any Open WebUI version):
        export OWUI_EMAIL="you@example.com"
        export OWUI_PASSWORD="yourpassword"

    Option B — API key (Admin Panel > Settings > Account > API Keys):
        export OWUI_API_KEY="sk-..."

Usage:
    python3 openweb_ui_push_tool.py              # push once
    python3 openweb_ui_push_tool.py --watch      # push automatically on every file save
    python3 openweb_ui_push_tool.py --probe      # discover working API endpoints (run this if push fails)
    python3 openweb_ui_push_tool.py --clean      # delete ALL versions of this tool then create one fresh copy
"""

import os
import re
import sys
import json
import time
import argparse
import subprocess
import requests

TOOL_FILE      = os.path.join(os.path.dirname(__file__), "openwebui_tool.py")
OWUI_URL       = os.getenv("OWUI_URL",        "http://localhost:3000")
OWUI_KEY       = os.getenv("OWUI_API_KEY",    "")
OWUI_EMAIL     = os.getenv("OWUI_EMAIL",      "")
OWUI_PASSWORD  = os.getenv("OWUI_PASSWORD",   "")
OWUI_CONTAINER = os.getenv("OWUI_CONTAINER",  "open-webui")


def get_token() -> str:
    """Return a Bearer token — from env API key, or by signing in with email/password."""
    if OWUI_KEY:
        return OWUI_KEY

    if OWUI_EMAIL and OWUI_PASSWORD:
        base = OWUI_URL.rstrip("/")
        try:
            resp = requests.post(
                f"{base}/api/v1/auths/signin",
                json={"email": OWUI_EMAIL, "password": OWUI_PASSWORD},
                timeout=10,
            )
            resp.raise_for_status()
            token = resp.json().get("token")
            if not token:
                print(f"Error: Sign-in succeeded but no token returned: {resp.text[:200]}")
                sys.exit(1)
            return token
        except requests.exceptions.HTTPError as e:
            print(f"Error: Login failed (HTTP {e.response.status_code}) — check OWUI_EMAIL and OWUI_PASSWORD")
            sys.exit(1)
        except requests.exceptions.ConnectionError:
            print(f"Error: Cannot connect to Open WebUI at {OWUI_URL}")
            sys.exit(1)

    print(
        "Error: No credentials set. Use one of:\n"
        "\n"
        "  Option A — email + password:\n"
        "    export OWUI_EMAIL='you@example.com'\n"
        "    export OWUI_PASSWORD='yourpassword'\n"
        "\n"
        "  Option B — API key:\n"
        "    export OWUI_API_KEY='sk-...'\n"
        "\n"
        "  Also set:  export OWUI_URL='http://your-server:3000'"
    )
    sys.exit(1)


def extract_meta(content: str) -> dict:
    def _get(field):
        m = re.search(rf"^{field}:\s*(.+)", content, re.MULTILINE)
        return m.group(1).strip() if m else ""

    title   = _get("title")       or "Web Search & URL Fetch"
    version = _get("version")     or "1.0.0"
    desc    = _get("description") or ""
    tool_id = re.sub(r"[^a-z0-9]+", "_", title.lower()).strip("_")
    return {"id": tool_id, "name": title, "version": version, "description": desc}


def _call(method: str, url: str, headers: dict, payload: dict) -> requests.Response:
    resp = requests.request(method, url, headers=headers, json=payload, timeout=10)
    return resp


def push(content: str, meta: dict) -> None:
    token   = get_token()
    headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
    base    = OWUI_URL.rstrip("/")

    resp = requests.get(f"{base}/api/v1/tools/", headers=headers, timeout=10)
    resp.raise_for_status()
    tools    = resp.json()
    existing = next(
        (t for t in tools if t.get("id") == meta["id"] or t.get("name") == meta["name"]),
        None,
    )

    payload = {
        "id":      meta["id"],
        "name":    meta["name"],
        "content": content,
        "meta": {
            "description": meta["description"],
            "manifest":    {"version": meta["version"]},
        },
    }

    if not existing:
        resp = _call("POST", f"{base}/api/v1/tools/create", headers, payload)
        if resp.ok:
            print(f"[{time.strftime('%H:%M:%S')}] Created  '{meta['name']}' (id={meta['id']})  v{meta['version']}")
            return
        raise RuntimeError(f"Create failed ({resp.status_code}): {resp.text[:200]}")

    eid = existing["id"]

    # Try API update verbs
    for method, url in [
        ("PUT",    f"{base}/api/v1/tools/{eid}"),
        ("PATCH",  f"{base}/api/v1/tools/{eid}"),
        ("POST",   f"{base}/api/v1/tools/{eid}/update"),
    ]:
        resp = _call(method, url, headers, payload)
        if resp.status_code not in (404, 405):
            resp.raise_for_status()
            print(f"[{time.strftime('%H:%M:%S')}] Updated  '{meta['name']}' (id={eid})  v{meta['version']}  [{method}]")
            return

    # Try delete + recreate
    for method, url in [
        ("DELETE", f"{base}/api/v1/tools/{eid}"),
        ("POST",   f"{base}/api/v1/tools/{eid}/delete"),
    ]:
        resp = _call(method, url, headers, {})
        if resp.status_code not in (404, 405):
            resp.raise_for_status()
            resp = _call("POST", f"{base}/api/v1/tools/create", headers, payload)
            resp.raise_for_status()
            print(f"[{time.strftime('%H:%M:%S')}] Updated  '{meta['name']}' (id={eid})  v{meta['version']}  [DELETE+CREATE]")
            return

    # Final fallback: update the database directly via docker exec
    push_via_docker(content, meta)


def push_via_docker(content: str, meta: dict) -> None:
    """Update the tool directly in Open WebUI's SQLite database via docker exec."""
    script = """
import sqlite3, json, os, glob, sys

content  = json.loads(sys.argv[1])
tool_id  = sys.argv[2]

candidates = [
    "/app/backend/data/webui.db",
    "/data/webui.db",
]
candidates += glob.glob("/app/**/*.db", recursive=True)

for db_path in candidates:
    if not os.path.exists(db_path):
        continue
    try:
        conn = sqlite3.connect(db_path)
        has_tool = conn.execute(
            "SELECT name FROM sqlite_master WHERE type='table' AND name='tool'"
        ).fetchone()
        if not has_tool:
            conn.close()
            continue
        conn.execute("UPDATE tool SET content=? WHERE id=?", (content, tool_id))
        rows = conn.execute("SELECT changes()").fetchone()[0]
        conn.commit()
        conn.close()
        if rows > 0:
            print("ok:" + db_path)
            sys.exit(0)
        else:
            print("no_rows:" + db_path)
    except Exception as e:
        print("err:" + db_path + ":" + str(e))

print("not_found")
sys.exit(1)
"""
    try:
        result = subprocess.run(
            ["docker", "exec", OWUI_CONTAINER, "python3", "-c", script,
             json.dumps(content), meta["id"]],
            capture_output=True,
            text=True,
            timeout=30,
        )
        output = result.stdout.strip()
        if result.returncode == 0 and output.startswith("ok:"):
            db_path = output[3:]
            print(f"[{time.strftime('%H:%M:%S')}] Updated  '{meta['name']}' (id={meta['id']})  v{meta['version']}  [docker→sqlite {db_path}]")
            return
        raise RuntimeError(output or result.stderr)
    except FileNotFoundError:
        raise RuntimeError("'docker' command not found — is Docker installed and on PATH?")
    except subprocess.TimeoutExpired:
        raise RuntimeError("docker exec timed out")
    except RuntimeError:
        raise
    except Exception as e:
        raise RuntimeError(f"docker exec failed: {e}") from e


def clean_and_push(content: str, meta: dict) -> None:
    """Delete every version of this tool from the DB by name match, then create one fresh copy."""
    token   = get_token()
    headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
    base    = OWUI_URL.rstrip("/")

    resp = requests.get(f"{base}/api/v1/tools/", headers=headers, timeout=10)
    resp.raise_for_status()
    all_tools = resp.json()

    target_name = meta["name"].lower()
    matches = [
        t for t in all_tools
        if target_name in t.get("name", "").lower()
        or t.get("id", "") == meta["id"]
    ]

    if matches:
        ids_to_delete = [t["id"] for t in matches]
        print(f"Found {len(matches)} existing tool(s) to remove:")
        for t in matches:
            print(f"  - '{t['name']}' (id={t['id']})")

        script = """
import sqlite3, json, os, glob, sys
ids = json.loads(sys.argv[1])
candidates = ["/app/backend/data/webui.db", "/data/webui.db"]
candidates += glob.glob("/app/**/*.db", recursive=True)
for db_path in candidates:
    if not os.path.exists(db_path):
        continue
    conn = sqlite3.connect(db_path)
    has = conn.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='tool'").fetchone()
    if not has:
        conn.close()
        continue
    for tid in ids:
        conn.execute("DELETE FROM tool WHERE id=?", (tid,))
    conn.commit()
    conn.close()
    print("deleted:" + db_path)
    sys.exit(0)
print("db_not_found")
sys.exit(1)
"""
        result = subprocess.run(
            ["docker", "exec", OWUI_CONTAINER, "python3", "-c", script,
             json.dumps(ids_to_delete)],
            capture_output=True, text=True, timeout=30,
        )
        out = result.stdout.strip()
        if result.returncode != 0 or not out.startswith("deleted:"):
            raise RuntimeError(f"Failed to delete from DB: {out or result.stderr}")
        print(f"Deleted from {out[len('deleted:'):]}")
    else:
        print("No existing versions found — creating fresh.")

    # Create the single clean copy
    payload = {
        "id":      meta["id"],
        "name":    meta["name"],
        "content": content,
        "meta": {
            "description": meta["description"],
            "manifest":    {"version": meta["version"]},
        },
    }
    resp = _call("POST", f"{base}/api/v1/tools/create", headers, payload)
    resp.raise_for_status()
    print(f"[{time.strftime('%H:%M:%S')}] Created  '{meta['name']}' (id={meta['id']})  v{meta['version']}  [clean install]")


def probe() -> None:
    """Try every known endpoint pattern and print the HTTP status for each."""
    token   = get_token()
    headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
    base    = OWUI_URL.rstrip("/")

    # Fetch real tool id to use in probes
    resp = requests.get(f"{base}/api/v1/tools/", headers=headers, timeout=10)
    resp.raise_for_status()
    tools = resp.json()
    eid   = tools[0]["id"] if tools else "test_id"
    dummy = {"id": eid, "name": "probe", "content": "", "meta": {}}

    print(f"Open WebUI: {base}")
    print(f"Using tool id for probe: {eid}")
    print(f"{'METHOD':<8} {'ENDPOINT':<55} STATUS")
    print("-" * 75)

    # Try to fetch the OpenAPI spec so we can see all real endpoints
    for spec_path in ("/openapi.json", "/docs/openapi.json", "/api/openapi.json"):
        try:
            r = requests.get(base + spec_path, headers=headers, timeout=6)
            if r.ok:
                paths = list(r.json().get("paths", {}).keys())
                tool_paths = [p for p in paths if "tool" in p.lower()]
                print(f"OpenAPI spec found at {spec_path}")
                print(f"  Tool-related paths: {tool_paths}")
                break
        except Exception:
            pass

    probe_urls = [
        # list
        ("GET",    f"/api/v1/tools/"),
        ("GET",    f"/api/tools/"),
        # read single
        ("GET",    f"/api/v1/tools/{eid}"),
        # update
        ("PUT",    f"/api/v1/tools/{eid}"),
        ("PATCH",  f"/api/v1/tools/{eid}"),
        ("POST",   f"/api/v1/tools/{eid}"),
        ("POST",   f"/api/v1/tools/{eid}/update"),
        ("PUT",    f"/api/v1/tools/id/{eid}"),
        ("POST",   f"/api/v1/tools/update"),
        # delete
        ("DELETE", f"/api/v1/tools/{eid}"),
        ("DELETE", f"/api/v1/tools/{eid}/delete"),
        ("POST",   f"/api/v1/tools/{eid}/delete"),
        ("DELETE", f"/api/tools/{eid}"),
        # create
        ("POST",   f"/api/v1/tools/"),
        ("POST",   f"/api/v1/tools/add"),
        ("POST",   f"/api/v1/tools/create"),
        ("PUT",    f"/api/v1/tools/"),
        ("POST",   f"/api/tools/"),
        ("POST",   f"/api/tools/add"),
    ]

    for method, path in probe_urls:
        url  = base + path
        try:
            r = requests.request(method, url, headers=headers, json=dummy, timeout=8)
            status = r.status_code
            allow  = r.headers.get("Allow", "")
            note   = ""
            if status < 300:
                note = " <-- WORKS"
            elif status == 401:
                note = " (auth error)"
            elif status == 404:
                note = " (not found)"
            elif status == 405:
                note = f" (wrong method — allowed: {allow})" if allow else " (wrong method)"
            elif status == 422:
                note = " (endpoint exists, payload rejected)"
            print(f"{method:<8} {path:<55} {status}{note}")
        except Exception as e:
            print(f"{method:<8} {path:<55} ERROR: {e}")

    print("\nShare the output above to identify which endpoint to use.")


def push_file() -> None:
    with open(TOOL_FILE, encoding="utf-8") as f:
        content = f.read()
    meta = extract_meta(content)
    push(content, meta)


def watch() -> None:
    print(f"Watching {TOOL_FILE} — press Ctrl+C to stop")
    last_mtime = None
    while True:
        try:
            mtime = os.path.getmtime(TOOL_FILE)
            if mtime != last_mtime:
                last_mtime = mtime
                if last_mtime is not None:
                    try:
                        push_file()
                    except Exception as e:
                        print(f"[{time.strftime('%H:%M:%S')}] Error: {e}")
            time.sleep(1)
        except KeyboardInterrupt:
            print("\nStopped.")
            break


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Push openwebui_tool.py to Open WebUI")
    parser.add_argument("--watch", action="store_true", help="Re-push on every file save")
    parser.add_argument("--probe", action="store_true", help="Discover working API endpoints")
    parser.add_argument("--clean", action="store_true", help="Delete all versions of this tool then create one fresh copy")
    args = parser.parse_args()

    if args.probe:
        probe()
    elif args.clean:
        with open(TOOL_FILE, encoding="utf-8") as f:
            content = f.read()
        meta = extract_meta(content)
        clean_and_push(content, meta)
    elif args.watch:
        watch()
    else:
        try:
            push_file()
        except requests.exceptions.ConnectionError:
            print(f"Error: Cannot connect to Open WebUI at {OWUI_URL}")
            sys.exit(1)
        except requests.exceptions.HTTPError as e:
            print(f"Error: HTTP {e.response.status_code} — {e.response.text[:200]}")
            sys.exit(1)