diff --git a/qortal_api_tool_function.py b/qortal_api_tool_function.py deleted file mode 100644 index d908dde..0000000 --- a/qortal_api_tool_function.py +++ /dev/null @@ -1,216 +0,0 @@ -""" -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 diff --git a/qortal_api_toolkit.py b/qortal_api_toolkit.py new file mode 100644 index 0000000..0fbbf30 --- /dev/null +++ b/qortal_api_toolkit.py @@ -0,0 +1,146 @@ +""" +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 'tools' undder 'workspaces' 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 +from typing import Optional +from pydantic import BaseModel, Field, ConfigDict + + +# ──────────────────────────────────────────────────────────── helpers ────── +class _EventEmitter: + def __init__(self, cb): # cb is Open WebUI’s internal emitter + 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-tweakable knobs appear in Settings → Tools + 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", + ) + HTTP_TIMEOUT: int = Field(default=30, description="Request timeout in seconds") + ALLOW_POST: bool = Field( + default=False, + description="Enable POST endpoints (node must allow them)", + ) + + def __init__(self): + self.valves = self.Valves() + self.headers = {"User-Agent": "Open-WebUI Qortal Toolkit (+https://qortal.org)"} + + # ─────────────────────────── single public function ──────────────────── + 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. + + • 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 + """ + emitter = _EventEmitter(__event_emitter__) + await emitter.emit(f"{method.upper()} {endpoint}") + + # parse the JSON strings the LLM supplies + try: + path_params_json = path_params_json or "{}" + query_params_json = query_params_json or "{}" + json_body_json = json_body_json or "{}" + 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 + 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: + 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 + + 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: + await emitter.emit(str(e), "error", True) + return json.dumps({"success": False, "error": str(e)})