diff --git a/qortal_api_pipeline.py b/qortal_api_pipeline.py index b4493be..80877ef 100644 --- a/qortal_api_pipeline.py +++ b/qortal_api_pipeline.py @@ -1,147 +1,323 @@ +""" +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_get`. + +________ __ .__ _____ __________.___ _____ __ .__ +\_____ \ ____________/ |______ | | / _ \\______ \ |/ ____\_ __ ____ _____/ |_|__| ____ ____ + / / \ \ / _ \_ __ \ __\__ \ | | / /_\ \| ___/ \ __\ | \/ \_/ ___\ __\ |/ _ \ / \ +/ \_/. ( <_> ) | \/| | / __ \| |_/ | \ | | || | | | / | \ \___| | | ( <_> ) | \ +\_____\ \_/\____/|__| |__| (____ /____|____|__ /____| |___||__| |____/|___| /\___ >__| |__|\____/|___| / + \__> \/ \/ \/ \/ \/ + + ___. __ .__ + \_ |__ ___.__. /\ ___________ ______ _ __ _____/ |_|__| ____ + | __ < | | \/ _/ ___\_ __ \/ _ \ \/ \/ // __ \ __\ |/ ___\ + | \_\ \___ | /\ \ \___| | \( <_> ) /\ ___/| | | \ \___ + |___ / ____| \/ \___ >__| \____/ \/\_/ \___ >__| |__|\___ > + \/\/ \/ \/ \/ +⭐ 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\"}"} -""" -title: Qortal API Pipeline -author: crowe & ChatGPT -git_url: https://github.com/crowetic/qortal-api-tool -description: Native Open WebUI tool that lets the LLM call any Qortal Core HTTP endpoint, - with automatic discovery and validation against the node's openapi.json. -required_open_webui_version: 0.5.0 -version: 0.2.0 -licence: MIT """ -import os -import json -import requests -from typing import Any, Dict, Optional, List +import os, json, re, requests +from typing import Optional +from pydantic import BaseModel, Field, ConfigDict -# --------------------------------------------------------------------------- -# Configuration -# --------------------------------------------------------------------------- -DEFAULT_QORTAL_URL = os.getenv("QORTAL_API_URL", "https://api.qortal.org").rstrip("/") -OPENAPI_URL = os.getenv("QORTAL_OPENAPI_URL", f"{DEFAULT_QORTAL_URL}/openapi.json") -# --------------------------------------------------------------------------- +# ───────────────────────────────────────────────────────── 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: - """ - ⚙️ Qortal API Toolkit for Open WebUI - ------------------------------------------------- - Generic wrappers + dynamic endpoint list pulled from openapi.json. - Methods exposed to the LLM: - • list_get_endpoints() – list of all available GET endpoints - • qortal_get() – perform validated HTTP GET - • qortal_post() – perform HTTP POST - """ + # ───────── user-visible knobs ───────── + class Valves(BaseModel): + model_config = ConfigDict(arbitrary_types_allowed=True) - # ---------------------- internal helpers ------------------------------ # - _openapi_cache: Optional[Dict[str, Any]] = None + 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)", + ) - def _build_url(self, endpoint: str, path_params: Optional[Dict[str, str]] = None) -> str: - """Replace {tokens} in endpoint and join with base URL.""" - cleaned = f"/{endpoint.lstrip('/')}" # ensure single leading slash - if path_params: - for k, v in path_params.items(): - cleaned = cleaned.replace("{" + k + "}", str(v)) - return DEFAULT_QORTAL_URL + cleaned + # ───────── 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 helpers ------------------------------- # - def _load_openapi(self) -> Dict[str, Any]: - if self._openapi_cache is not None: - return self._openapi_cache + # ───────── 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" + ) try: - resp = requests.get(OPENAPI_URL, timeout=30) - resp.raise_for_status() - self._openapi_cache = resp.json() - except Exception as e: - self._openapi_cache = {} - return self._openapi_cache + r = requests.get( + url, headers=self.headers, timeout=self.valves.HTTP_TIMEOUT + ) + r.raise_for_status() + self._spec_cache = r.json() + except Exception: + self._spec_cache = {} + return self._spec_cache - def list_get_endpoints(self) -> List[str]: - """Return all GET endpoints available on this Qortal node.""" - spec = self._load_openapi() - paths = spec.get("paths", {}) - return [p for p, verbs in paths.items() if "get" in verbs] - - def _is_valid_get(self, endpoint: str) -> bool: - endpoint = "/" + endpoint.lstrip("/") - return endpoint in self.list_get_endpoints() - - # -------------------------- request core ------------------------------ # - def _request( + # ───────── one-shot helper: discover → describe → GET ───────── + async def qortal_auto_get( self, - method: str, - endpoint: str, - path_params: Optional[Dict[str, Any]] = None, - query_params: Optional[Dict[str, Any]] = None, - json_body: Optional[Dict[str, Any]] = None, - validate_get: bool = True, - ) -> Dict[str, Any]: - if method.upper() == "GET" and validate_get and not self._is_valid_get(endpoint): - return { - "success": False, - "error": f"Endpoint '{endpoint}' is not listed as GET in node's OpenAPI spec", - "url": None, - } + path: str, + path_params_json: str = "{}", + query_params_json: str = "{}", + __event_emitter__=None, + ) -> str: + """ + 🔹 ONE call does it all 🔹 + 1. Verifies the path exists in openapi.json + 2. Fills any {tokens} from path_params_json (errors if missing) + 3. Performs the GET and returns the JSON - url = self._build_url(endpoint, path_params) + Example: + {"path": "/names/{name}", + "path_params_json": "{\"name\":\"crowetic\"}"} + """ + emitter = _EventEmitter(__event_emitter__) + + # 1) make sure the path is valid + spec = await self._load_spec() + if path not in spec.get("paths", {}): + await emitter.emit("Unknown path", "error", True) + return json.dumps({"success": False, "error": "unknown_path"}) + + # 2) fill tokens + path_params_json = path_params_json or "{}" try: - resp = requests.request( - method=method.upper(), - url=url, + path_params = json.loads(path_params_json) + except json.JSONDecodeError as e: + await emitter.emit("Bad JSON in path_params_json", "error", True) + return json.dumps({"success": False, "error": str(e)}) + + missing = [ + t[1:-1] for t in re.findall(r"{[^}]+}", path) if t[1:-1] not in path_params + ] + if missing: + await emitter.emit("Missing path params", "error", True) + return json.dumps( + {"success": False, "error": "missing_path_params", "needs": missing} + ) + + # 3) forward to the normal call wrapper + return await self.qortal_call( + endpoint=path, + method="GET", + path_params_json=path_params_json, + query_params_json=query_params_json or "{}", + json_body_json="{}", + __event_emitter__=__event_emitter__, + ) + + # ───────── 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, + } + ) + + # ───────── main call functions ───────── + async def qortal_get_simple( + self, + endpoint: str, + path_params_json: str = "{}", + query_params_json: str = "{}", + __event_emitter__=None, + ) -> str: + """ + Convenience wrapper for simple GETs. + Same behaviour as qortal_call but the method is locked to GET. + """ + return await self.qortal_call( + endpoint=endpoint, + method="GET", + path_params_json=path_params_json or "{}", + query_params_json=query_params_json or "{}", + json_body_json="{}", + __event_emitter__=__event_emitter__, + ) + + 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 + try: + r = requests.request( + method.upper(), + url, params=query_params, - json=json_body, - timeout=30, + json=json_body if method.upper() == "POST" else None, + headers=self.headers, + timeout=self.valves.HTTP_TIMEOUT, ) try: - data = resp.json() + data = r.json() except ValueError: - data = resp.text - return { - "status_code": resp.status_code, - "success": resp.ok, - "url": resp.url, - "data": data, - } + data = r.text + + await emitter.emit(f"→ {r.status_code}", "complete", True) + return json.dumps( + { + "success": r.ok, + "status_code": r.status_code, + "url": r.url, + "data": data, + }, + ensure_ascii=False, + ) except Exception as e: - return {"success": False, "error": str(e), "url": url} - - # -------------------------- PUBLIC TOOLS ----------------------------- # - def qortal_get( - self, - endpoint: str, - path_params: Optional[dict] = None, - query_params: Optional[dict] = None, - ) -> dict: - """ - Generic HTTP GET to Qortal Core. - - Parameters - ---------- - endpoint : str - Endpoint path (e.g. "/addresses/balance/{address}") - path_params : dict, optional - Dict for replacement of {tokens} in endpoint. - query_params : dict, optional - Dict for URL query parameters. - """ - return self._request("GET", endpoint, path_params, query_params) - - def qortal_post( - self, - endpoint: str, - path_params: Optional[dict] = None, - json_body: Optional[dict] = None, - ) -> dict: - """ - Generic HTTP POST to Qortal Core. - - Parameters - ---------- - endpoint : str - Endpoint path (e.g. "/arbitrary/publish") - path_params : dict, optional - Dict for replacement of {tokens} in endpoint. - json_body : dict, optional - Dict to send as JSON body. - """ - return self._request("POST", endpoint, path_params, None, json_body, validate_get=False) + await emitter.emit(str(e), "error", True) + return json.dumps({"success": False, "error": str(e)}) diff --git a/qortal_api_toolkit.py b/qortal_api_toolkit.py index 0fbbf30..83a4b07 100644 --- a/qortal_api_toolkit.py +++ b/qortal_api_toolkit.py @@ -8,12 +8,12 @@ description: Query any Qortal Core HTTP endpoint directly from Open WebUI. HOW-TO ------ -1. Add this in 'tools' undder 'workspaces' to create the tool function +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_get`. + The LLM will auto-call `qortal_orchestrate`. ________ __ .__ _____ __________.___ _____ __ .__ \_____ \ ____________/ |______ | | / _ \\______ \ |/ ____\_ __ ____ _____/ |_|__| ____ ____ @@ -28,16 +28,28 @@ ________ __ .__ _____ __________.___ _____ | \_\ \___ | /\ \ \___| | \( <_> ) /\ ___/| | | \ \___ |___ / ____| \/ \___ >__| \____/ \/\_/ \___ >__| |__|\___ > \/\/ \/ \/ \/ +⭐ 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, requests +import os, json, re, requests, difflib, asyncio from typing import Optional from pydantic import BaseModel, Field, ConfigDict -# ──────────────────────────────────────────────────────────── helpers ────── +# ───────────────────────────────────────────────────────── helpers ───────── class _EventEmitter: - def __init__(self, cb): # cb is Open WebUI’s internal emitter + def __init__(self, cb): self.cb = cb async def emit(self, msg, status="in_progress", done=False): @@ -54,9 +66,9 @@ class _EventEmitter: ) -# ──────────────────────────────────────────────────────────── TOOL ───────── +# ───────────────────────────────────────────────────────── TOOL ──────────── class Tools: - # user-tweakable knobs appear in Settings → Tools + # ───────── user-visible knobs ───────── class Valves(BaseModel): model_config = ConfigDict(arbitrary_types_allowed=True) @@ -64,17 +76,87 @@ class Tools: default=os.getenv("QORTAL_API_URL", "https://api.qortal.org"), description="Base URL of your Qortal node", ) - HTTP_TIMEOUT: int = Field(default=30, description="Request timeout in seconds") + 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, + } + ) - # ─────────────────────────── single public function ──────────────────── async def qortal_call( self, endpoint: str, @@ -87,36 +169,62 @@ class Tools: """ Generic Qortal HTTP call. - • endpoint (str) – e.g. "/addresses/balance/{address}" - • method (str) – "GET" or "POST" - • path_params_json (str) – JSON dict of tokens to replace - • query_params_json (str) – JSON dict for URL ?key=value - • json_body_json (str) – JSON dict for POST body + 🟢 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 the JSON strings the LLM supplies + # parse JSON args try: - path_params_json = path_params_json or "{}" - query_params_json = query_params_json or "{}" - json_body_json = json_body_json or "{}" + 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)}) - # build full URL + # 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)) - # block POST if valve is off 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"}) - # make the request - try: + # perform request + def do_request(): r = requests.request( method.upper(), url, @@ -129,18 +237,201 @@ class Tools: data = r.json() except ValueError: data = r.text + return { + "ok": r.ok, + "status": r.status_code, + "url": r.url, + "data": data, + } - await emitter.emit(f"→ {r.status_code}", "complete", True) - return json.dumps( - { - "success": r.ok, - "status_code": r.status_code, - "url": r.url, - "data": data, - }, - ensure_ascii=False, - ) - + 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 emitter.emit(str(e), "error", True) + 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, + } diff --git a/setup-ai-stack.sh b/setup-ai-stack.sh index 809c9c5..9571bfe 100644 --- a/setup-ai-stack.sh +++ b/setup-ai-stack.sh @@ -79,7 +79,7 @@ if [[ "$1" == "--update" ]]; then --restart always \ ghcr.io/open-webui/open-webui:ollama else - echo "pipelines not running, skipping update" + echo "open-webui not running, skipping update" fi # localai (GPU-support) @@ -478,6 +478,7 @@ if [[ "$1" != "--update" ]]; then -e OPENAI_API_BASE_URL=http://pipelines:9099 \ -e OPENAI_API_KEY=0p3n-w3bu! \ -e OLLAMA_MAX_LOADED_MODELS=2 \ + -e OLLAMA_BASE_URL=http://localhost:11434 \ -v ollama:/root/.ollama \ -v open-webui:/app/backend/data \ --name open-webui \