438 lines
17 KiB
Python
438 lines
17 KiB
Python
"""
|
||
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,
|
||
}
|