|
# retoor <retoor@molodetz.nl>
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import re
|
|
import uuid
|
|
from typing import Any, Awaitable, Callable
|
|
|
|
from devplacepy.services.devii.errors import ToolInputError
|
|
from devplacepy.services.devii.virtual_tools.store import VirtualToolStore
|
|
|
|
logger = logging.getLogger("devii.virtual_tools.controller")
|
|
|
|
NAME_PATTERN = re.compile(r"^[a-zA-Z][a-zA-Z0-9_]{1,40}$")
|
|
PREVIEW_CHARS = 200
|
|
|
|
Evaluator = Callable[[str], Awaitable[str]]
|
|
|
|
|
|
def _serialize(row: dict[str, Any], full: bool = False) -> dict[str, Any]:
|
|
data = {
|
|
"name": row.get("name"),
|
|
"description": row.get("description"),
|
|
"input_description": row.get("input_description") or "",
|
|
"enabled": bool(row.get("enabled", 1)),
|
|
"updated_at": row.get("updated_at"),
|
|
}
|
|
if full:
|
|
data["prompt"] = row.get("prompt") or ""
|
|
else:
|
|
data["prompt_preview"] = (row.get("prompt") or "")[:PREVIEW_CHARS]
|
|
return data
|
|
|
|
|
|
class VirtualToolController:
|
|
def __init__(self, store: VirtualToolStore, evaluator: Evaluator, builtin_names: set[str]) -> None:
|
|
self._store = store
|
|
self._evaluator = evaluator
|
|
self._builtin_names = builtin_names
|
|
|
|
async def dispatch(self, name: str, arguments: dict[str, Any]) -> str:
|
|
handlers = {
|
|
"tool_create": self._create,
|
|
"tool_list": self._list,
|
|
"tool_get": self._get,
|
|
"tool_update": self._update,
|
|
"tool_delete": self._delete,
|
|
}
|
|
handler = handlers.get(name)
|
|
if handler is None:
|
|
raise ToolInputError(f"Unknown virtual-tool action: {name}")
|
|
return handler(arguments)
|
|
|
|
def has(self, name: str) -> bool:
|
|
row = self._store.find(name)
|
|
return bool(row and row.get("enabled", 1))
|
|
|
|
async def run(self, name: str, arguments: dict[str, Any]) -> str:
|
|
row = self._store.find(name)
|
|
if not row or not row.get("enabled", 1):
|
|
raise ToolInputError(f"No enabled virtual tool named '{name}'.")
|
|
user_input = str(arguments.get("input", "")).strip()
|
|
prompt = row.get("prompt") or ""
|
|
composed = f"{prompt}\n\nUser input: {user_input}" if user_input else prompt
|
|
result = await self._evaluator(composed)
|
|
return json.dumps({"status": "success", "tool": name, "result": result}, ensure_ascii=False)
|
|
|
|
def _require_name(self, arguments: dict[str, Any]) -> str:
|
|
name = str(arguments.get("name", "")).strip()
|
|
if not name:
|
|
raise ToolInputError("A tool 'name' is required.")
|
|
return name
|
|
|
|
def _create(self, arguments: dict[str, Any]) -> str:
|
|
name = self._require_name(arguments)
|
|
if not NAME_PATTERN.match(name):
|
|
raise ToolInputError(
|
|
"Tool name must be 2-41 characters: a letter followed by letters, digits, or underscores."
|
|
)
|
|
if name in self._builtin_names:
|
|
raise ToolInputError(f"'{name}' is a built-in tool name; choose a different name.")
|
|
if self._store.find(name):
|
|
raise ToolInputError(f"A tool named '{name}' already exists; use tool_update to change it.")
|
|
description = str(arguments.get("description", "")).strip()
|
|
prompt = str(arguments.get("prompt", "")).strip()
|
|
if not description:
|
|
raise ToolInputError("A 'description' is required so the tool can be triggered correctly.")
|
|
if not prompt:
|
|
raise ToolInputError("A 'prompt' is required: the instruction the tool runs when called.")
|
|
record = {
|
|
"uid": uuid.uuid4().hex,
|
|
"name": name,
|
|
"description": description,
|
|
"prompt": prompt,
|
|
"input_description": str(arguments.get("input_description", "")).strip(),
|
|
"enabled": 1,
|
|
}
|
|
stored = self._store.create(record)
|
|
return json.dumps({"status": "created", "tool": _serialize(stored)}, ensure_ascii=False)
|
|
|
|
def _list(self, arguments: dict[str, Any]) -> str:
|
|
rows = self._store.list()
|
|
return json.dumps(
|
|
{"count": len(rows), "tools": [_serialize(row) for row in rows]}, ensure_ascii=False
|
|
)
|
|
|
|
def _get(self, arguments: dict[str, Any]) -> str:
|
|
name = self._require_name(arguments)
|
|
row = self._store.find(name)
|
|
if not row:
|
|
raise ToolInputError(f"No tool named '{name}'.")
|
|
return json.dumps({"tool": _serialize(row, full=True)}, ensure_ascii=False)
|
|
|
|
def _update(self, arguments: dict[str, Any]) -> str:
|
|
name = self._require_name(arguments)
|
|
row = self._store.find(name)
|
|
if not row:
|
|
raise ToolInputError(f"No tool named '{name}'.")
|
|
changes: dict[str, Any] = {}
|
|
if "description" in arguments and str(arguments["description"]).strip():
|
|
changes["description"] = str(arguments["description"]).strip()
|
|
if "prompt" in arguments and str(arguments["prompt"]).strip():
|
|
changes["prompt"] = str(arguments["prompt"]).strip()
|
|
if "input_description" in arguments:
|
|
changes["input_description"] = str(arguments["input_description"]).strip()
|
|
if "enabled" in arguments and arguments["enabled"] is not None:
|
|
changes["enabled"] = 1 if str(arguments["enabled"]).strip().lower() in ("1", "true", "yes", "on") else 0
|
|
if not changes:
|
|
raise ToolInputError("No updatable fields supplied (description, prompt, input_description, enabled).")
|
|
self._store.update(name, changes)
|
|
return json.dumps({"status": "updated", "tool": _serialize(self._store.find(name))}, ensure_ascii=False)
|
|
|
|
def _delete(self, arguments: dict[str, Any]) -> str:
|
|
name = self._require_name(arguments)
|
|
deleted = self._store.delete(name)
|
|
if not deleted:
|
|
raise ToolInputError(f"No tool named '{name}'.")
|
|
return json.dumps({"status": "deleted", "name": name}, ensure_ascii=False)
|