| 1 | """Base class for agent tools.""" |
| 2 | |
| 3 | from abc import ABC, abstractmethod |
| 4 | from collections.abc import Callable |
| 5 | from copy import deepcopy |
| 6 | from typing import Any, TypeVar |
| 7 | |
| 8 | _ToolT = TypeVar("_ToolT", bound="Tool") |
| 9 | |
| 10 | # Matches :meth:`Tool._cast_value` / :meth:`Schema.validate_json_schema_value` behavior |
| 11 | _JSON_TYPE_MAP: dict[str, type | tuple[type, ...]] = { |
| 12 | "string": str, |
| 13 | "integer": int, |
| 14 | "number": (int, float), |
| 15 | "boolean": bool, |
| 16 | "array": list, |
| 17 | "object": dict, |
| 18 | } |
| 19 | |
| 20 | |
| 21 | class Schema(ABC): |
| 22 | """Abstract base for JSON Schema fragments describing tool parameters. |
| 23 | |
| 24 | Concrete types live in :mod:`nanobot.agent.tools.schema`; all implement |
| 25 | :meth:`to_json_schema` and :meth:`validate_value`. Class methods |
| 26 | :meth:`validate_json_schema_value` and :meth:`fragment` are the shared validation and normalization entry points. |
| 27 | """ |
| 28 | |
| 29 | @staticmethod |
| 30 | def resolve_json_schema_type(t: Any) -> str | None: |
| 31 | """Resolve the non-null type name from JSON Schema ``type`` (e.g. ``['string','null']`` -> ``'string'``).""" |
| 32 | if isinstance(t, list): |
| 33 | return next((x for x in t if x != "null"), None) |
| 34 | return t # type: ignore[return-value] |
| 35 | |
| 36 | @staticmethod |
| 37 | def subpath(path: str, key: str) -> str: |
| 38 | return f"{path}.{key}" if path else key |
| 39 | |
| 40 | @staticmethod |
| 41 | def validate_json_schema_value(val: Any, schema: dict[str, Any], path: str = "") -> list[str]: |
| 42 | """Validate ``val`` against a JSON Schema fragment; returns error messages (empty means valid). |
| 43 | |
| 44 | Used by :class:`Tool` and each concrete Schema's :meth:`validate_value`. |
| 45 | """ |
| 46 | raw_type = schema.get("type") |
| 47 | nullable = (isinstance(raw_type, list) and "null" in raw_type) or schema.get("nullable", False) |
| 48 | t = Schema.resolve_json_schema_type(raw_type) |
| 49 | label = path or "parameter" |
| 50 | |
| 51 | if nullable and val is None: |
| 52 | return [] |
| 53 | if t == "integer" and (not isinstance(val, int) or isinstance(val, bool)): |
| 54 | return [f"{label} should be integer"] |
| 55 | if t == "number" and ( |
| 56 | not isinstance(val, _JSON_TYPE_MAP["number"]) or isinstance(val, bool) |
| 57 | ): |
| 58 | return [f"{label} should be number"] |
| 59 | if t in _JSON_TYPE_MAP and t not in ("integer", "number") and not isinstance(val, _JSON_TYPE_MAP[t]): |
| 60 | return [f"{label} should be {t}"] |
| 61 | |
| 62 | errors: list[str] = [] |
| 63 | if "enum" in schema and val not in schema["enum"]: |
| 64 | errors.append(f"{label} must be one of {schema['enum']}") |
| 65 | if t in ("integer", "number"): |
| 66 | if "minimum" in schema and val < schema["minimum"]: |
| 67 | errors.append(f"{label} must be >= {schema['minimum']}") |
| 68 | if "maximum" in schema and val > schema["maximum"]: |
| 69 | errors.append(f"{label} must be <= {schema['maximum']}") |
| 70 | if t == "string": |
| 71 | if "minLength" in schema and len(val) < schema["minLength"]: |
| 72 | errors.append(f"{label} must be at least {schema['minLength']} chars") |
| 73 | if "maxLength" in schema and len(val) > schema["maxLength"]: |
| 74 | errors.append(f"{label} must be at most {schema['maxLength']} chars") |
| 75 | if t == "object": |
| 76 | props = schema.get("properties", {}) |
| 77 | for k in schema.get("required", []): |
| 78 | if k not in val: |
| 79 | errors.append(f"missing required {Schema.subpath(path, k)}") |
| 80 | for k, v in val.items(): |
| 81 | if k in props: |
| 82 | errors.extend(Schema.validate_json_schema_value(v, props[k], Schema.subpath(path, k))) |
| 83 | if t == "array": |
| 84 | if "minItems" in schema and len(val) < schema["minItems"]: |
| 85 | errors.append(f"{label} must have at least {schema['minItems']} items") |
| 86 | if "maxItems" in schema and len(val) > schema["maxItems"]: |
| 87 | errors.append(f"{label} must be at most {schema['maxItems']} items") |
| 88 | if "items" in schema: |
| 89 | prefix = f"{path}[{{}}]" if path else "[{}]" |
| 90 | for i, item in enumerate(val): |
| 91 | errors.extend( |
| 92 | Schema.validate_json_schema_value(item, schema["items"], prefix.format(i)) |
| 93 | ) |
| 94 | return errors |
| 95 | |
| 96 | @staticmethod |
| 97 | def fragment(value: Any) -> dict[str, Any]: |
| 98 | """Normalize a Schema instance or an existing JSON Schema dict to a fragment dict.""" |
| 99 | # Try to_json_schema first: Schema instances must be distinguished from dicts that are already JSON Schema |
| 100 | to_js = getattr(value, "to_json_schema", None) |
| 101 | if callable(to_js): |
| 102 | return to_js() |
| 103 | if isinstance(value, dict): |
| 104 | return value |
| 105 | raise TypeError(f"Expected schema object or dict, got {type(value).__name__}") |
| 106 | |
| 107 | @abstractmethod |
| 108 | def to_json_schema(self) -> dict[str, Any]: |
| 109 | """Return a fragment dict compatible with :meth:`validate_json_schema_value`.""" |
| 110 | ... |
| 111 | |
| 112 | def validate_value(self, value: Any, path: str = "") -> list[str]: |
| 113 | """Validate a single value; returns error messages (empty means pass). Subclasses may override for extra rules.""" |
| 114 | return Schema.validate_json_schema_value(value, self.to_json_schema(), path) |
| 115 | |
| 116 | |
| 117 | class Tool(ABC): |
| 118 | """Agent capability: read files, run commands, etc.""" |
| 119 | |
| 120 | _TYPE_MAP = { |
| 121 | "string": str, |
| 122 | "integer": int, |
| 123 | "number": (int, float), |
| 124 | "boolean": bool, |
| 125 | "array": list, |
| 126 | "object": dict, |
| 127 | } |
| 128 | _BOOL_TRUE = frozenset(("true", "1", "yes")) |
| 129 | _BOOL_FALSE = frozenset(("false", "0", "no")) |
| 130 | |
| 131 | @staticmethod |
| 132 | def _resolve_type(t: Any) -> str | None: |
| 133 | """Pick first non-null type from JSON Schema unions like ``['string','null']``.""" |
| 134 | return Schema.resolve_json_schema_type(t) |
| 135 | |
| 136 | @property |
| 137 | @abstractmethod |
| 138 | def name(self) -> str: |
| 139 | """Tool name used in function calls.""" |
| 140 | ... |
| 141 | |
| 142 | @property |
| 143 | @abstractmethod |
| 144 | def description(self) -> str: |
| 145 | """Description of what the tool does.""" |
| 146 | ... |
| 147 | |
| 148 | @property |
| 149 | @abstractmethod |
| 150 | def parameters(self) -> dict[str, Any]: |
| 151 | """JSON Schema for tool parameters.""" |
| 152 | ... |
| 153 | |
| 154 | @property |
| 155 | def read_only(self) -> bool: |
| 156 | """Whether this tool is side-effect free and safe to parallelize.""" |
| 157 | return False |
| 158 | |
| 159 | @property |
| 160 | def concurrency_safe(self) -> bool: |
| 161 | """Whether this tool can run alongside other concurrency-safe tools.""" |
| 162 | return self.read_only and not self.exclusive |
| 163 | |
| 164 | @property |
| 165 | def exclusive(self) -> bool: |
| 166 | """Whether this tool should run alone even if concurrency is enabled.""" |
| 167 | return False |
| 168 | |
| 169 | @abstractmethod |
| 170 | async def execute(self, **kwargs: Any) -> Any: |
| 171 | """Run the tool; returns a string or list of content blocks.""" |
| 172 | ... |
| 173 | |
| 174 | def _cast_object(self, obj: Any, schema: dict[str, Any]) -> dict[str, Any]: |
| 175 | if not isinstance(obj, dict): |
| 176 | return obj |
| 177 | props = schema.get("properties", {}) |
| 178 | return {k: self._cast_value(v, props[k]) if k in props else v for k, v in obj.items()} |
| 179 | |
| 180 | def cast_params(self, params: dict[str, Any]) -> dict[str, Any]: |
| 181 | """Apply safe schema-driven casts before validation.""" |
| 182 | schema = self.parameters or {} |
| 183 | if schema.get("type", "object") != "object": |
| 184 | return params |
| 185 | return self._cast_object(params, schema) |
| 186 | |
| 187 | def _cast_value(self, val: Any, schema: dict[str, Any]) -> Any: |
| 188 | t = self._resolve_type(schema.get("type")) |
| 189 | |
| 190 | if t == "boolean" and isinstance(val, bool): |
| 191 | return val |
| 192 | if t == "integer" and isinstance(val, int) and not isinstance(val, bool): |
| 193 | return val |
| 194 | if t in self._TYPE_MAP and t not in ("boolean", "integer", "array", "object"): |
| 195 | expected = self._TYPE_MAP[t] |
| 196 | if isinstance(val, expected): |
| 197 | return val |
| 198 | |
| 199 | if isinstance(val, str) and t in ("integer", "number"): |
| 200 | try: |
| 201 | return int(val) if t == "integer" else float(val) |
| 202 | except ValueError: |
| 203 | return val |
| 204 | |
| 205 | if t == "string": |
| 206 | return val if val is None else str(val) |
| 207 | |
| 208 | if t == "boolean" and isinstance(val, str): |
| 209 | low = val.lower() |
| 210 | if low in self._BOOL_TRUE: |
| 211 | return True |
| 212 | if low in self._BOOL_FALSE: |
| 213 | return False |
| 214 | return val |
| 215 | |
| 216 | if t == "array" and isinstance(val, list): |
| 217 | items = schema.get("items") |
| 218 | return [self._cast_value(x, items) for x in val] if items else val |
| 219 | |
| 220 | if t == "object" and isinstance(val, dict): |
| 221 | return self._cast_object(val, schema) |
| 222 | |
| 223 | return val |
| 224 | |
| 225 | def validate_params(self, params: dict[str, Any]) -> list[str]: |
| 226 | """Validate against JSON schema; empty list means valid.""" |
| 227 | if not isinstance(params, dict): |
| 228 | return [f"parameters must be an object, got {type(params).__name__}"] |
| 229 | schema = self.parameters or {} |
| 230 | if schema.get("type", "object") != "object": |
| 231 | raise ValueError(f"Schema must be object type, got {schema.get('type')!r}") |
| 232 | return Schema.validate_json_schema_value(params, {**schema, "type": "object"}, "") |
| 233 | |
| 234 | def to_schema(self) -> dict[str, Any]: |
| 235 | """OpenAI function schema.""" |
| 236 | return { |
| 237 | "type": "function", |
| 238 | "function": { |
| 239 | "name": self.name, |
| 240 | "description": self.description, |
| 241 | "parameters": self.parameters, |
| 242 | }, |
| 243 | } |
| 244 | |
| 245 | |
| 246 | def tool_parameters(schema: dict[str, Any]) -> Callable[[type[_ToolT]], type[_ToolT]]: |
| 247 | """Class decorator: attach JSON Schema and inject a concrete ``parameters`` property. |
| 248 | |
| 249 | Use on ``Tool`` subclasses instead of writing ``@property def parameters``. The |
| 250 | schema is stored on the class and returned as a fresh copy on each access. |
| 251 | |
| 252 | Example:: |
| 253 | |
| 254 | @tool_parameters({ |
| 255 | "type": "object", |
| 256 | "properties": {"path": {"type": "string"}}, |
| 257 | "required": ["path"], |
| 258 | }) |
| 259 | class ReadFileTool(Tool): |
| 260 | ... |
| 261 | """ |
| 262 | |
| 263 | def decorator(cls: type[_ToolT]) -> type[_ToolT]: |
| 264 | frozen = deepcopy(schema) |
| 265 | |
| 266 | @property |
| 267 | def parameters(self: Any) -> dict[str, Any]: |
| 268 | return deepcopy(frozen) |
| 269 | |
| 270 | cls._tool_parameters_schema = deepcopy(frozen) |
| 271 | cls.parameters = parameters # type: ignore[assignment] |
| 272 | |
| 273 | abstract = getattr(cls, "__abstractmethods__", None) |
| 274 | if abstract is not None and "parameters" in abstract: |
| 275 | cls.__abstractmethods__ = frozenset(abstract - {"parameters"}) # type: ignore[misc] |
| 276 | |
| 277 | return cls |
| 278 | |
| 279 | return decorator |
| 280 |