324 lines
12 KiB
Python
324 lines
12 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_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\"}"}
|
||
|
||
"""
|
||
|
||
import os, json, re, requests
|
||
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"
|
||
)
|
||
try:
|
||
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
|
||
|
||
# ───────── one-shot helper: discover → describe → GET ─────────
|
||
async def qortal_auto_get(
|
||
self,
|
||
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
|
||
|
||
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:
|
||
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 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)})
|