From 61c97a5cb2c52f59ac58dc637d02b58d7155b891 Mon Sep 17 00:00:00 2001 From: crowetic Date: Wed, 21 May 2025 13:41:03 -0700 Subject: [PATCH] added test qortal api tool function for open-webui --- qortal_api_tool.py => qortal_api_pipeline.py | 0 qortal_api_tool_function.py | 216 +++++++++++++++++++ 2 files changed, 216 insertions(+) rename qortal_api_tool.py => qortal_api_pipeline.py (100%) create mode 100644 qortal_api_tool_function.py diff --git a/qortal_api_tool.py b/qortal_api_pipeline.py similarity index 100% rename from qortal_api_tool.py rename to qortal_api_pipeline.py diff --git a/qortal_api_tool_function.py b/qortal_api_tool_function.py new file mode 100644 index 0000000..d908dde --- /dev/null +++ b/qortal_api_tool_function.py @@ -0,0 +1,216 @@ +""" +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`. + +________ __ .__ _____ __________.___ _____ __ .__ +\_____ \ ____________/ |______ | | / _ \\______ \ |/ ____\_ __ ____ _____/ |_|__| ____ ____ + / / \ \ / _ \_ __ \ __\__ \ | | / /_\ \| ___/ \ __\ | \/ \_/ ___\ __\ |/ _ \ / \ +/ \_/. ( <_> ) | \/| | / __ \| |_/ | \ | | || | | | / | \ \___| | | ( <_> ) | \ +\_____\ \_/\____/|__| |__| (____ /____|____|__ /____| |___||__| |____/|___| /\___ >__| |__|\____/|___| / + \__> \/ \/ \/ \/ \/ + + ___. __ .__ + \_ |__ ___.__. /\ ___________ ______ _ __ _____/ |_|__| ____ + | __ < | | \/ _/ ___\_ __ \/ _ \ \/ \/ // __ \ __\ |/ ___\ + | \_\ \___ | /\ \ \___| | \( <_> ) /\ ___/| | | \ \___ + |___ / ____| \/ \___ >__| \____/ \/\_/ \___ >__| |__|\___ > + \/\/ \/ \/ \/ +""" + +import os, json, requests, asyncio +from typing import Dict, Any, Optional, Callable +from pydantic import BaseModel, Field + + +# ╭─────────────────────────── helpers ───────────────────────────╮ +class _EventEmitter: + def __init__(self, cb: Callable[[dict], Any] | None): + self.cb = cb + + async def emit(self, description, status="in_progress", done=False): + if self.cb: + await self.cb( + { + "type": "status", + "data": { + "status": status, + "description": description, + "done": done, + }, + } + ) + + +# ╭─────────────────────────── plugin ────────────────────────────╮ +class Tools: + # ─── user-visible knobs ────────────────────────────────────── + class Valves(BaseModel): + QORTAL_URL: str = Field( + default=os.getenv("QORTAL_API_URL", "https://api.qortal.org"), + description="Base URL of your Qortal node (http[s]://host:port)", + ) + OPENAPI_URL: str = Field( + default=os.getenv("QORTAL_OPENAPI_URL", ""), + description="Custom location of openapi.json (leave blank for /openapi.json)", + ) + HTTP_TIMEOUT: int = Field(default=30, description="Request timeout (seconds)") + ALLOW_POST: bool = Field( + default=False, + description="Enable POST endpoints (requires unlocked API on the node)", + ) + + def __init__(self): + self.valves = self.Valves() + self._openapi_cache: Optional[Dict[str, Any]] = None + self.headers = { + "User-Agent": "Open-WebUI Qortal API Toolkit (+https://qortal.org)" + } + + # ─── public async methods (tool functions) ─────────────────── + async def qortal_get( + self, + endpoint: str, + path_params: Dict[str, str] | None = None, + query_params: Dict[str, str] | None = None, + __event_emitter__: Callable[[dict], Any] | None = None, + ) -> str: + """ + Perform a validated **GET** request to the node. + + • endpoint – e.g. "/addresses/balance/{address}" + • path_params – dict that fills the {tokens} + • query_params – optional ?key=value pairs + """ + emitter = _EventEmitter(__event_emitter__) + await emitter.emit(f"GET {endpoint}") + + if not await self._is_valid_get(endpoint): + await emitter.emit( + f"Endpoint '{endpoint}' is not a valid GET per node spec", "error", True + ) + return json.dumps({"error": "invalid_endpoint"}) + + return await self._do_request( + "GET", endpoint, path_params, query_params, None, emitter + ) + + async def qortal_post( + self, + endpoint: str, + path_params: Dict[str, str] | None = None, + json_body: Dict[str, Any] | None = None, + __event_emitter__: Callable[[dict], Any] | None = None, + ) -> str: + """ + Perform a **POST** request (requires ALLOW_POST = true). + + • endpoint – e.g. "/arbitrary/publish" + • path_params – dict that fills the {tokens} + • json_body – payload sent as JSON + """ + emitter = _EventEmitter(__event_emitter__) + await emitter.emit(f"POST {endpoint}") + + if not self.valves.ALLOW_POST: + await emitter.emit( + "POST disabled in valves; set ALLOW_POST = true to enable", + "error", + True, + ) + return json.dumps({"error": "post_disabled"}) + + return await self._do_request( + "POST", endpoint, path_params, None, json_body, emitter + ) + + # ╭──────────────────── internal plumbing ───────────────────╮ + async def _do_request( + self, + method: str, + endpoint: str, + path_params: Dict[str, Any] | None, + query_params: Dict[str, Any] | None, + json_body: Dict[str, Any] | None, + emitter: _EventEmitter, + ) -> str: + url = self._build_url(endpoint, path_params) + + try: + resp = requests.request( + method=method, + url=url, + params=query_params, + json=json_body, + headers=self.headers, + timeout=self.valves.HTTP_TIMEOUT, + ) + data: Any + try: + data = resp.json() + except ValueError: + data = resp.text + + await emitter.emit( + f"{method} {endpoint} → {resp.status_code}", "complete", True + ) + return json.dumps( + { + "success": resp.ok, + "status_code": resp.status_code, + "url": resp.url, + "data": data, + }, + ensure_ascii=False, + ) + + except Exception as e: + await emitter.emit(f"Request error: {e}", "error", True) + return json.dumps({"success": False, "error": str(e)}) + + # ─── spec helpers ─────────────────────────────────────────── + async def _is_valid_get(self, endpoint: str) -> bool: + endpoint = "/" + endpoint.lstrip("/") + return endpoint in await self._valid_get_list() + + async def _valid_get_list(self): + spec = await self._load_openapi() + return [ + p for p, verbs in spec.get("paths", {}).items() if "get" in verbs + ] + + async def _load_openapi(self): + if self._openapi_cache is not None: + return self._openapi_cache + + url = ( + self.valves.OPENAPI_URL + or f"{self.valves.QORTAL_URL.rstrip('/')}/openapi.json" + ) + try: + resp = requests.get(url, headers=self.headers, timeout=self.valves.HTTP_TIMEOUT) + resp.raise_for_status() + self._openapi_cache = resp.json() + except Exception: + self._openapi_cache = {} + return self._openapi_cache + + # ─── misc helpers ─────────────────────────────────────────── + def _build_url(self, endpoint: str, path_params: Dict[str, Any] | None): + ep = "/" + endpoint.lstrip("/") + if path_params: + for k, v in path_params.items(): + ep = ep.replace("{" + k + "}", str(v)) + return self.valves.QORTAL_URL.rstrip("/") + ep