AI-Dev/qortal_api_toolkit.py

"""
title: Qortal API Toolkit
author: crowetic (with assistance from chatgpt)
git_url: https://gitea.qortal.link/crowetic/AI-Dev
version: 0.3.0
license: MIT
description: Query any Qortal Core HTTP endpoint directly from Open WebUI.

HOW-TO
------
1. Add this in 'functions' to create the tool function
2. Add the tool to your model
3. In any model/preset: *Chat → ⚙️ Tools* → enable **Qortal API Toolkit**.
4. Ask something like:
     “Get the balance of address Q...”
   The LLM will auto-call `qortal_orchestrate`.

________                 __         .__     _____ __________.___  _____                    __  .__               
\_____  \   ____________/  |______  |  |   /  _  \\______   \   |/ ____\_ __  ____   _____/  |_|__| ____   ____  
 /  / \  \ /  _ \_  __ \   __\__  \ |  |  /  /_\  \|     ___/   \   __\  |  \/    \_/ ___\   __\  |/  _ \ /    \ 
/   \_/.  (  <_> )  | \/|  |  / __ \|  |_/    |    \    |   |   ||  | |  |  /   |  \  \___|  | |  (  <_> )   |  \
\_____\ \_/\____/|__|   |__| (____  /____|____|__  /____|   |___||__| |____/|___|  /\___  >__| |__|\____/|___|  /
        \__>                       \/             \/                              \/     \/                    \/

                                            ___.                                                 __  .__        
                                    \_ |__ ___.__. /\   ___________  ______  _  __ _____/  |_|__| ____  
                                    | __ <   |  | \/ _/ ___\_  __ \/  _ \ \/ \/ // __ \   __\  |/ ___\ 
                                    | \_\ \___  | /\ \  \___|  | \(  <_> )     /\  ___/|  | |  \  \___ 
                                    |___  / ____| \/  \___  >__|   \____/ \/\_/  \___  >__| |__|\___  >
                                        \/\/              \/                         \/             \/  
⭐ Quick examples ⭐
• Get block height
    {"endpoint":"/blocks/height"}

• Get balance
    {"endpoint":"/addresses/balance/{address}",
     "path_params_json":"{\"address\":\"Q...\"}"}

• Publish (POST)
    {"endpoint":"/arbitrary/publish","method":"POST",
     "json_body_json":"{\"name\":\"example\",\"service\":\"SCRIPT\"}"}

"""

import os, json, re, requests, difflib, asyncio
from typing import Optional
from pydantic import BaseModel, Field, ConfigDict


# ───────────────────────────────────────────────────────── helpers ─────────
class _EventEmitter:
    def __init__(self, cb):
        self.cb = cb

    async def emit(self, msg, status="in_progress", done=False):
        if self.cb:
            await self.cb(
                {
                    "type": "status",
                    "data": {
                        "status": status,
                        "description": msg,
                        "done": done,
                    },
                }
            )


# ───────────────────────────────────────────────────────── TOOL ────────────
class Tools:
    # ───────── user-visible knobs ─────────
    class Valves(BaseModel):
        model_config = ConfigDict(arbitrary_types_allowed=True)

        QORTAL_URL: str = Field(
            default=os.getenv("QORTAL_API_URL", "https://api.qortal.org"),
            description="Base URL of your Qortal node",
        )
        OPENAPI_URL: str = Field(
            default=os.getenv("QORTAL_OPENAPI_URL", ""),
            description="Custom openapi.json URL (leave blank for default)",
        )
        HTTP_TIMEOUT: int = Field(default=30, description="Request timeout (s)")
        ALLOW_POST: bool = Field(
            default=False,
            description="Enable POST endpoints (node must allow them)",
        )

    # ───────── initialiser ─────────
    def __init__(self):
        self.valves = self.Valves()
        self.headers = {"User-Agent": "Open-WebUI Qortal Toolkit (+https://qortal.org)"}
        self._spec_cache = None  # cache for openapi.json

    # ───────── OpenAPI loader ─────────
    async def _load_spec(self):
        if self._spec_cache is not None:
            return self._spec_cache

        url = (
            self.valves.OPENAPI_URL
            or f"{self.valves.QORTAL_URL.rstrip('/')}/openapi.json"
        )

        def fetch():
            r = requests.get(url, headers=self.headers, timeout=self.valves.HTTP_TIMEOUT)
            r.raise_for_status()
            return r.json()

        loop = asyncio.get_running_loop()
        try:
            self._spec_cache = await loop.run_in_executor(None, fetch)
        except Exception:
            self._spec_cache = {}
        return self._spec_cache


    # ───────── discovery helpers ─────────
    async def qortal_openapi(self) -> str:
        """Return raw openapi.json."""
        return json.dumps(await self._load_spec())

    async def qortal_list_paths(self) -> str:
        """List every path in the spec."""
        return json.dumps(list((await self._load_spec()).get("paths", {}).keys()))

    async def qortal_describe(self, path: str) -> str:
        """Return the schema entry for a single path."""
        return json.dumps(
            (await self._load_spec()).get("paths", {}).get(path)
            or {"error": "unknown_path"}
        )

    async def qortal_build_call(
        self, path: str, example_values_json: str = "{}"
    ) -> str:
        """
        Return a skeleton argument dict for qortal_call.
        Utilize this to figure out the details of an endpoint prior to making the call.
        """
        spec = await self._load_spec()
        if path not in spec.get("paths", {}):
            return json.dumps({"error": "unknown_path"})

        example_vals = json.loads(example_values_json or "{}")
        tokens = [seg[1:-1] for seg in path.split("/") if seg.startswith("{")]
        missing = [t for t in tokens if t not in example_vals]

        return json.dumps(
            {
                "endpoint": path,
                "method": "GET",
                "path_params_json": json.dumps(example_vals or {}),
                "query_params_json": "{}",
                "json_body_json": "{}",
                "needs": missing,
            }
        )

    async def qortal_call(
        self,
        endpoint: str,
        method: str = "GET",
        path_params_json: str = "{}",
        query_params_json: str = "{}",
        json_body_json: str = "{}",
        __event_emitter__=None,
    ) -> str:
        """
        Generic Qortal HTTP call.

        🟢 Correct template
        {
          "endpoint": "/names/{name}",
          "method":   "GET",               <-- string, default "GET"
          "path_params_json":  "{\"name\":\"crowetic\"}",
          "query_params_json": "{}",
          "json_body_json":  "{}"
        }

        🔴 Common mistakes (DON’T DO THESE)
        {
          "endpoint": "GET",               <-- wrong field, value swapped
          "path": "/names/{name}"          <-- 'path' is not a parameter
        }
        """
        path_params_json = path_params_json or "{}"
        query_params_json = query_params_json or "{}"
        json_body_json = json_body_json or "{}"

        # swap-correction guard
        if endpoint.upper() in ("GET", "POST") and method.startswith("/"):
            endpoint, method = method, endpoint  # silently swap them

        emitter = _EventEmitter(__event_emitter__)
        await emitter.emit(f"{method.upper()} {endpoint}")

        # parse JSON args
        try:
            path_params = json.loads(path_params_json)
            query_params = json.loads(query_params_json)
            json_body = json.loads(json_body_json)
        except json.JSONDecodeError as e:
            await emitter.emit("Bad JSON in arguments", "error", True)
            return json.dumps({"success": False, "error": str(e)})

        # check missing tokens
        unfilled = [
            tok for tok in re.findall(r"{(.*?)}", endpoint) if tok not in path_params
        ]
        if unfilled:
            await emitter.emit("Missing path parameters", "error", True)
            return json.dumps(
                {"success": False, "error": "missing_path_params", "needs": unfilled}
            )

        # build URL
        url = self.valves.QORTAL_URL.rstrip("/") + "/" + endpoint.lstrip("/")
        for k, v in path_params.items():
            url = url.replace("{" + k + "}", str(v))

        if method.upper() == "POST" and not self.valves.ALLOW_POST:
            await emitter.emit("POST disabled", "error", True)
            return json.dumps({"success": False, "error": "post_disabled"})

        # perform request
        def do_request():
            r = requests.request(
                method.upper(),
                url,
                params=query_params,
                json=json_body if method.upper() == "POST" else None,
                headers=self.headers,
                timeout=self.valves.HTTP_TIMEOUT,
            )
            try:
                data = r.json()
            except ValueError:
                data = r.text
            return {
                "ok":        r.ok,
                "status":    r.status_code,
                "url":       r.url,
                "data":      data,
            }

        loop = asyncio.get_running_loop()
        try:
            resp = await loop.run_in_executor(None, do_request)
            await _EventEmitter(__event_emitter__).emit(f"→ {resp['status']}", "complete", True)
            return json.dumps({
                "success":     resp["ok"],
                "status_code": resp["status"],
                "url":         resp["url"],
                "data":        resp["data"],
            }, ensure_ascii=False)
        except Exception as e:
            await _EventEmitter(__event_emitter__).emit(str(e), "error", True)
            return json.dumps({"success": False, "error": str(e)})

    def extract_key_points(self, user_query: str) -> dict:
        """
        Heuristic parser that returns a dict:
          - action:   e.g. "get balance", "query block height"
          - addresses: list of detected Qortal addresses
          - params:   other parsed params like height or name
        """
        q = user_query.strip()

        # 1) Detect all Qortal addresses (Q + 27–34 alphanum chars)
        addresses = re.findall(r"\bQ[a-zA-Z0-9]{26,33}\b", q)

        # 2) Detect numeric params (height, limit, etc.)
        params = {}
        m = re.search(r"\bheight\s*(?:is|=|:)?\s*(\d+)\b", q, flags=re.IGNORECASE)
        if m:
            params["height"] = int(m.group(1))

        m = re.search(r"\bname\s*(?:is|=|:)?\s*([A-Za-z0-9_]+)\b", q, flags=re.IGNORECASE)
        if m:
            params["name"] = m.group(1)

        m = re.search(r"\blimit\s*(?:is|=|:)?\s*(\d+)\b", q, flags=re.IGNORECASE)
        if m:
            params["limit"] = int(m.group(1))

        # 3) Fallback: any other key: value pairs “key: value”
        for key, val in re.findall(r"(\w+)\s*:\s*([^\s,]+)", q):
            if key.lower() not in params:
                # try casting to int
                if val.isdigit():
                    params[key.lower()] = int(val)
                else:
                    params[key.lower()] = val

        # 4) Extract an “action” phrase: 
        #    verb (get|query|fetch) + everything up until keyword “of”/address/param
        action = ""
        m = re.search(
            r"(?i)\b(get|query|fetch)\s+(.+?)(?=\s+(of\b|\bQ[a-zA-Z0-9]{26,33}\b|\bheight\b|\bname\b|$))",
            q,
        )
        if m:
            action = f"{m.group(1).lower()} {m.group(2).strip().lower()}"
        else:
            # fallback: first 4 words
            action = " ".join(q.split()[:4]).lower()

        return {
            "action": action,
            "addresses": addresses,
            "params": params,
        }

    def find_candidate_endpoints(self, key_points: dict, limit: int) -> list[str]:
        """
        Fuzzy-match key words (e.g. 'balance', 'height') against each path summary.
        """
        spec = self._spec_cache or {}
        paths = spec.get("paths", {})
        summaries = {
            p: (ops.get("get", {}).get("summary", "") + " " +
                ops.get("post", {}).get("summary", ""))
            for p, ops in paths.items()
        }
        # pick the top `limit` matches against the user query keywords
        pool = list(summaries.keys())
        # flatten key words
        keywords = []
        if key_points["action"]:
            keywords += key_points["action"].split()        # split action into words
        keywords += key_points["addresses"]                  # each address
        for k, v in key_points["params"].items():
            keywords.append(k)                               # param name
            keywords.append(str(v))                          # param value

        scores = {}
        for p in pool:
            text = summaries[p].lower() + " " + p.lower()
            scores[p] = max(difflib.SequenceMatcher(None, text, kw.lower()).ratio()
                            for kw in keywords) if keywords else 0
        ranked = sorted(scores, key=lambda p: scores[p], reverse=True)
        return ranked[:limit]

    async def get_endpoint_schema(self, path: str) -> dict:
        spec = await self._load_spec()
        return spec.get("paths", {}).get(path, {})

    def build_call_args(
        self,
        path: str,
        schema: dict,
        key_points: dict,
        attempt: int
    ) -> dict:
        """
        Fill in path/query/body based on key_points.
        You could randomize or tweak defaults per attempt.
        """
        # get required path tokens
        tokens = [seg[1:-1] for seg in path.split("/") if seg.startswith("{")]
        # naive: take first address or number for each
        path_vals = {}
        for token in tokens:
            if token == "address" and key_points["addresses"]:
                path_vals[token] = key_points["addresses"][0]
            elif token in key_points["params"]:
                path_vals[token] = key_points["params"][token]
        return {
            "endpoint": path,
            "method": "GET" if "get" in schema else "POST",
            "path_params_json": json.dumps(path_vals),
            "query_params_json": "{}",
            "json_body_json": "{}",
        }

    def is_good_response(self, resp: dict) -> bool:
        """
        Decide whether resp['data'] looks valid.
        """
        if not resp.get("success", False):
            return False
        data = resp.get("data")
        # simple heuristics: non-empty dict or list
        return bool(data and (isinstance(data, dict) or isinstance(data, list)))

    async def orchestrate_call(
        self,
        user_query: str,
        max_calls: int = 3,
        max_attempts_per_call: int = 4,
        __event_emitter__=None,
    ) -> str:
        emitter = _EventEmitter(__event_emitter__)
        key_points = self.extract_key_points(user_query)
        spec = await self._load_spec()

        candidates = self.find_candidate_endpoints(key_points, max_calls)
        last_resp = {"success": False, "error": "no_attempts"}
        for path in candidates:
            schema = await self.get_endpoint_schema(path)
            await emitter.emit(f"Trying endpoint {path}", "in_progress")

            for attempt in range(1, max_attempts_per_call + 1):
                args = self.build_call_args(path, schema, key_points, attempt)
                resp = await self.qortal_call(__event_emitter__=__event_emitter__, **args)
                resp_obj = json.loads(resp)
                if self.is_good_response(resp_obj):
                    await emitter.emit(
                        f"Success on {path} attempt {attempt}", "complete", True
                    )
                    return json.dumps(resp_obj, ensure_ascii=False)
                last_resp = resp_obj
                await emitter.emit(
                    f"Attempt {attempt} failed for {path}", "error", False
                )
            await emitter.emit(f"Moving to next endpoint", "in_progress")

        # after all attempts
        await emitter.emit("All attempts exhausted; please clarify.", "error", True)
        return json.dumps(last_resp, ensure_ascii=False)

toolkit = Tools()

def register_workspace_tools() -> dict:
    """
    Open-WebUI will call this to learn about your functions.
    Keys are the “tool names” the model invokes, values are the callables.
    """
    return {
        "qortal_openapi":    toolkit.qortal_openapi,
        "qortal_list_paths": toolkit.qortal_list_paths,
        "qortal_describe":   toolkit.qortal_describe,
        "qortal_build_call": toolkit.qortal_build_call,
        "qortal_call":       toolkit.qortal_call,
        "qortal_orchestrate":toolkit.orchestrate_call,
    }