crowetic/AI-Dev
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

resolved issues with the open-webui tool

Browse Source
This commit is contained in:
crowetic 2025-05-21 14:24:32 -07:00
parent 61c97a5cb2
commit d33d899d5f
2 changed files with 146 additions and 216 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

216
qortal_api_tool_function.py
View File

@ -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 <QORTAL_URL>/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

146
qortal_api_toolkit.py Normal file
View File

@ -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)})