diff --git a/nanobot/agent/tools/__init__.py b/nanobot/agent/tools/__init__.py index aac5d7d91..c005cc6b5 100644 --- a/nanobot/agent/tools/__init__.py +++ b/nanobot/agent/tools/__init__.py @@ -1,6 +1,27 @@ """Agent tools module.""" -from nanobot.agent.tools.base import Tool +from nanobot.agent.tools.base import Schema, Tool, tool_parameters from nanobot.agent.tools.registry import ToolRegistry +from nanobot.agent.tools.schema import ( + ArraySchema, + BooleanSchema, + IntegerSchema, + NumberSchema, + ObjectSchema, + StringSchema, + tool_parameters_schema, +) -__all__ = ["Tool", "ToolRegistry"] +__all__ = [ + "Schema", + "ArraySchema", + "BooleanSchema", + "IntegerSchema", + "NumberSchema", + "ObjectSchema", + "StringSchema", + "Tool", + "ToolRegistry", + "tool_parameters", + "tool_parameters_schema", +] diff --git a/nanobot/agent/tools/base.py b/nanobot/agent/tools/base.py index f119f6908..5e19e5c40 100644 --- a/nanobot/agent/tools/base.py +++ b/nanobot/agent/tools/base.py @@ -1,16 +1,120 @@ """Base class for agent tools.""" from abc import ABC, abstractmethod -from typing import Any +from collections.abc import Callable +from typing import Any, TypeVar + +_ToolT = TypeVar("_ToolT", bound="Tool") + +# Matches :meth:`Tool._cast_value` / :meth:`Schema.validate_json_schema_value` behavior +_JSON_TYPE_MAP: dict[str, type | tuple[type, ...]] = { + "string": str, + "integer": int, + "number": (int, float), + "boolean": bool, + "array": list, + "object": dict, +} + + +class Schema(ABC): + """Abstract base for JSON Schema fragments describing tool parameters. + + Concrete types live in :mod:`nanobot.agent.tools.schema`; all implement + :meth:`to_json_schema` and :meth:`validate_value`. Class methods + :meth:`validate_json_schema_value` and :meth:`fragment` are the shared validation and normalization entry points. + """ + + @staticmethod + def resolve_json_schema_type(t: Any) -> str | None: + """Resolve the non-null type name from JSON Schema ``type`` (e.g. ``['string','null']`` -> ``'string'``).""" + if isinstance(t, list): + return next((x for x in t if x != "null"), None) + return t # type: ignore[return-value] + + @staticmethod + def subpath(path: str, key: str) -> str: + return f"{path}.{key}" if path else key + + @staticmethod + def validate_json_schema_value(val: Any, schema: dict[str, Any], path: str = "") -> list[str]: + """Validate ``val`` against a JSON Schema fragment; returns error messages (empty means valid). + + Used by :class:`Tool` and each concrete Schema's :meth:`validate_value`. + """ + raw_type = schema.get("type") + nullable = (isinstance(raw_type, list) and "null" in raw_type) or schema.get("nullable", False) + t = Schema.resolve_json_schema_type(raw_type) + label = path or "parameter" + + if nullable and val is None: + return [] + if t == "integer" and (not isinstance(val, int) or isinstance(val, bool)): + return [f"{label} should be integer"] + if t == "number" and ( + not isinstance(val, _JSON_TYPE_MAP["number"]) or isinstance(val, bool) + ): + return [f"{label} should be number"] + if t in _JSON_TYPE_MAP and t not in ("integer", "number") and not isinstance(val, _JSON_TYPE_MAP[t]): + return [f"{label} should be {t}"] + + errors: list[str] = [] + if "enum" in schema and val not in schema["enum"]: + errors.append(f"{label} must be one of {schema['enum']}") + if t in ("integer", "number"): + if "minimum" in schema and val < schema["minimum"]: + errors.append(f"{label} must be >= {schema['minimum']}") + if "maximum" in schema and val > schema["maximum"]: + errors.append(f"{label} must be <= {schema['maximum']}") + if t == "string": + if "minLength" in schema and len(val) < schema["minLength"]: + errors.append(f"{label} must be at least {schema['minLength']} chars") + if "maxLength" in schema and len(val) > schema["maxLength"]: + errors.append(f"{label} must be at most {schema['maxLength']} chars") + if t == "object": + props = schema.get("properties", {}) + for k in schema.get("required", []): + if k not in val: + errors.append(f"missing required {Schema.subpath(path, k)}") + for k, v in val.items(): + if k in props: + errors.extend(Schema.validate_json_schema_value(v, props[k], Schema.subpath(path, k))) + if t == "array": + if "minItems" in schema and len(val) < schema["minItems"]: + errors.append(f"{label} must have at least {schema['minItems']} items") + if "maxItems" in schema and len(val) > schema["maxItems"]: + errors.append(f"{label} must be at most {schema['maxItems']} items") + if "items" in schema: + prefix = f"{path}[{{}}]" if path else "[{}]" + for i, item in enumerate(val): + errors.extend( + Schema.validate_json_schema_value(item, schema["items"], prefix.format(i)) + ) + return errors + + @staticmethod + def fragment(value: Any) -> dict[str, Any]: + """Normalize a Schema instance or an existing JSON Schema dict to a fragment dict.""" + # Try to_json_schema first: Schema instances must be distinguished from dicts that are already JSON Schema + to_js = getattr(value, "to_json_schema", None) + if callable(to_js): + return to_js() + if isinstance(value, dict): + return value + raise TypeError(f"Expected schema object or dict, got {type(value).__name__}") + + @abstractmethod + def to_json_schema(self) -> dict[str, Any]: + """Return a fragment dict compatible with :meth:`validate_json_schema_value`.""" + ... + + def validate_value(self, value: Any, path: str = "") -> list[str]: + """Validate a single value; returns error messages (empty means pass). Subclasses may override for extra rules.""" + return Schema.validate_json_schema_value(value, self.to_json_schema(), path) class Tool(ABC): - """ - Abstract base class for agent tools. - - Tools are capabilities that the agent can use to interact with - the environment, such as reading files, executing commands, etc. - """ + """Agent capability: read files, run commands, etc.""" _TYPE_MAP = { "string": str, @@ -20,38 +124,31 @@ class Tool(ABC): "array": list, "object": dict, } + _BOOL_TRUE = frozenset(("true", "1", "yes")) + _BOOL_FALSE = frozenset(("false", "0", "no")) @staticmethod def _resolve_type(t: Any) -> str | None: - """Resolve JSON Schema type to a simple string. - - JSON Schema allows ``"type": ["string", "null"]`` (union types). - We extract the first non-null type so validation/casting works. - """ - if isinstance(t, list): - for item in t: - if item != "null": - return item - return None - return t + """Pick first non-null type from JSON Schema unions like ``['string','null']``.""" + return Schema.resolve_json_schema_type(t) @property @abstractmethod def name(self) -> str: """Tool name used in function calls.""" - pass + ... @property @abstractmethod def description(self) -> str: """Description of what the tool does.""" - pass + ... @property @abstractmethod def parameters(self) -> dict[str, Any]: """JSON Schema for tool parameters.""" - pass + ... @property def read_only(self) -> bool: @@ -70,142 +167,71 @@ class Tool(ABC): @abstractmethod async def execute(self, **kwargs: Any) -> Any: - """ - Execute the tool with given parameters. + """Run the tool; returns a string or list of content blocks.""" + ... - Args: - **kwargs: Tool-specific parameters. - - Returns: - Result of the tool execution (string or list of content blocks). - """ - pass + def _cast_object(self, obj: Any, schema: dict[str, Any]) -> dict[str, Any]: + if not isinstance(obj, dict): + return obj + props = schema.get("properties", {}) + return {k: self._cast_value(v, props[k]) if k in props else v for k, v in obj.items()} def cast_params(self, params: dict[str, Any]) -> dict[str, Any]: """Apply safe schema-driven casts before validation.""" schema = self.parameters or {} if schema.get("type", "object") != "object": return params - return self._cast_object(params, schema) - def _cast_object(self, obj: Any, schema: dict[str, Any]) -> dict[str, Any]: - """Cast an object (dict) according to schema.""" - if not isinstance(obj, dict): - return obj - - props = schema.get("properties", {}) - result = {} - - for key, value in obj.items(): - if key in props: - result[key] = self._cast_value(value, props[key]) - else: - result[key] = value - - return result - def _cast_value(self, val: Any, schema: dict[str, Any]) -> Any: - """Cast a single value according to schema.""" - target_type = self._resolve_type(schema.get("type")) + t = self._resolve_type(schema.get("type")) - if target_type == "boolean" and isinstance(val, bool): + if t == "boolean" and isinstance(val, bool): return val - if target_type == "integer" and isinstance(val, int) and not isinstance(val, bool): + if t == "integer" and isinstance(val, int) and not isinstance(val, bool): return val - if target_type in self._TYPE_MAP and target_type not in ("boolean", "integer", "array", "object"): - expected = self._TYPE_MAP[target_type] + if t in self._TYPE_MAP and t not in ("boolean", "integer", "array", "object"): + expected = self._TYPE_MAP[t] if isinstance(val, expected): return val - if target_type == "integer" and isinstance(val, str): + if isinstance(val, str) and t in ("integer", "number"): try: - return int(val) + return int(val) if t == "integer" else float(val) except ValueError: return val - if target_type == "number" and isinstance(val, str): - try: - return float(val) - except ValueError: - return val - - if target_type == "string": + if t == "string": return val if val is None else str(val) - if target_type == "boolean" and isinstance(val, str): - val_lower = val.lower() - if val_lower in ("true", "1", "yes"): + if t == "boolean" and isinstance(val, str): + low = val.lower() + if low in self._BOOL_TRUE: return True - if val_lower in ("false", "0", "no"): + if low in self._BOOL_FALSE: return False return val - if target_type == "array" and isinstance(val, list): - item_schema = schema.get("items") - return [self._cast_value(item, item_schema) for item in val] if item_schema else val + if t == "array" and isinstance(val, list): + items = schema.get("items") + return [self._cast_value(x, items) for x in val] if items else val - if target_type == "object" and isinstance(val, dict): + if t == "object" and isinstance(val, dict): return self._cast_object(val, schema) return val def validate_params(self, params: dict[str, Any]) -> list[str]: - """Validate tool parameters against JSON schema. Returns error list (empty if valid).""" + """Validate against JSON schema; empty list means valid.""" if not isinstance(params, dict): return [f"parameters must be an object, got {type(params).__name__}"] schema = self.parameters or {} if schema.get("type", "object") != "object": raise ValueError(f"Schema must be object type, got {schema.get('type')!r}") - return self._validate(params, {**schema, "type": "object"}, "") - - def _validate(self, val: Any, schema: dict[str, Any], path: str) -> list[str]: - raw_type = schema.get("type") - nullable = (isinstance(raw_type, list) and "null" in raw_type) or schema.get( - "nullable", False - ) - t, label = self._resolve_type(raw_type), path or "parameter" - if nullable and val is None: - return [] - if t == "integer" and (not isinstance(val, int) or isinstance(val, bool)): - return [f"{label} should be integer"] - if t == "number" and ( - not isinstance(val, self._TYPE_MAP[t]) or isinstance(val, bool) - ): - return [f"{label} should be number"] - if t in self._TYPE_MAP and t not in ("integer", "number") and not isinstance(val, self._TYPE_MAP[t]): - return [f"{label} should be {t}"] - - errors = [] - if "enum" in schema and val not in schema["enum"]: - errors.append(f"{label} must be one of {schema['enum']}") - if t in ("integer", "number"): - if "minimum" in schema and val < schema["minimum"]: - errors.append(f"{label} must be >= {schema['minimum']}") - if "maximum" in schema and val > schema["maximum"]: - errors.append(f"{label} must be <= {schema['maximum']}") - if t == "string": - if "minLength" in schema and len(val) < schema["minLength"]: - errors.append(f"{label} must be at least {schema['minLength']} chars") - if "maxLength" in schema and len(val) > schema["maxLength"]: - errors.append(f"{label} must be at most {schema['maxLength']} chars") - if t == "object": - props = schema.get("properties", {}) - for k in schema.get("required", []): - if k not in val: - errors.append(f"missing required {path + '.' + k if path else k}") - for k, v in val.items(): - if k in props: - errors.extend(self._validate(v, props[k], path + "." + k if path else k)) - if t == "array" and "items" in schema: - for i, item in enumerate(val): - errors.extend( - self._validate(item, schema["items"], f"{path}[{i}]" if path else f"[{i}]") - ) - return errors + return Schema.validate_json_schema_value(params, {**schema, "type": "object"}, "") def to_schema(self) -> dict[str, Any]: - """Convert tool to OpenAI function schema format.""" + """OpenAI function schema.""" return { "type": "function", "function": { @@ -214,3 +240,39 @@ class Tool(ABC): "parameters": self.parameters, }, } + + +def tool_parameters(schema: dict[str, Any]) -> Callable[[type[_ToolT]], type[_ToolT]]: + """Class decorator: attach JSON Schema and inject a concrete ``parameters`` property. + + Use on ``Tool`` subclasses instead of writing ``@property def parameters``. The + schema is stored on the class (shallow-copied) as ``_tool_parameters_schema``. + + Example:: + + @tool_parameters({ + "type": "object", + "properties": {"path": {"type": "string"}}, + "required": ["path"], + }) + class ReadFileTool(Tool): + ... + """ + + def decorator(cls: type[_ToolT]) -> type[_ToolT]: + frozen = dict(schema) + + @property + def parameters(self: Any) -> dict[str, Any]: + return frozen + + cls._tool_parameters_schema = frozen + cls.parameters = parameters # type: ignore[assignment] + + abstract = getattr(cls, "__abstractmethods__", None) + if abstract is not None and "parameters" in abstract: + cls.__abstractmethods__ = frozenset(abstract - {"parameters"}) # type: ignore[misc] + + return cls + + return decorator diff --git a/nanobot/agent/tools/cron.py b/nanobot/agent/tools/cron.py index ada55d7cf..064b6e4c9 100644 --- a/nanobot/agent/tools/cron.py +++ b/nanobot/agent/tools/cron.py @@ -4,11 +4,37 @@ from contextvars import ContextVar from datetime import datetime from typing import Any -from nanobot.agent.tools.base import Tool +from nanobot.agent.tools.base import Tool, tool_parameters +from nanobot.agent.tools.schema import BooleanSchema, IntegerSchema, StringSchema, tool_parameters_schema from nanobot.cron.service import CronService from nanobot.cron.types import CronJob, CronJobState, CronSchedule +@tool_parameters( + tool_parameters_schema( + action=StringSchema("Action to perform", enum=["add", "list", "remove"]), + message=StringSchema( + "Instruction for the agent to execute when the job triggers " + "(e.g., 'Send a reminder to WeChat: xxx' or 'Check system status and report')" + ), + every_seconds=IntegerSchema(0, description="Interval in seconds (for recurring tasks)"), + cron_expr=StringSchema("Cron expression like '0 9 * * *' (for scheduled tasks)"), + tz=StringSchema( + "Optional IANA timezone for cron expressions (e.g. 'America/Vancouver'). " + "When omitted with cron_expr, the tool's default timezone applies." + ), + at=StringSchema( + "ISO datetime for one-time execution (e.g. '2026-02-12T10:30:00'). " + "Naive values use the tool's default timezone." + ), + deliver=BooleanSchema( + description="Whether to deliver the execution result to the user channel (default true)", + default=True, + ), + job_id=StringSchema("Job ID (for remove)"), + required=["action"], + ) +) class CronTool(Tool): """Tool to schedule reminders and recurring tasks.""" @@ -64,49 +90,6 @@ class CronTool(Tool): f"If tz is omitted, cron expressions and naive ISO times default to {self._default_timezone}." ) - @property - def parameters(self) -> dict[str, Any]: - return { - "type": "object", - "properties": { - "action": { - "type": "string", - "enum": ["add", "list", "remove"], - "description": "Action to perform", - }, - "message": {"type": "string", "description": "Instruction for the agent to execute when the job triggers (e.g., 'Send a reminder to WeChat: xxx' or 'Check system status and report')"}, - "every_seconds": { - "type": "integer", - "description": "Interval in seconds (for recurring tasks)", - }, - "cron_expr": { - "type": "string", - "description": "Cron expression like '0 9 * * *' (for scheduled tasks)", - }, - "tz": { - "type": "string", - "description": ( - "Optional IANA timezone for cron expressions " - f"(e.g. 'America/Vancouver'). Defaults to {self._default_timezone}." - ), - }, - "at": { - "type": "string", - "description": ( - "ISO datetime for one-time execution " - f"(e.g. '2026-02-12T10:30:00'). Naive values default to {self._default_timezone}." - ), - }, - "deliver": { - "type": "boolean", - "description": "Whether to deliver the execution result to the user channel (default true)", - "default": True - }, - "job_id": {"type": "string", "description": "Job ID (for remove)"}, - }, - "required": ["action"], - } - async def execute( self, action: str, diff --git a/nanobot/agent/tools/filesystem.py b/nanobot/agent/tools/filesystem.py index e3a8fecaf..11f05c557 100644 --- a/nanobot/agent/tools/filesystem.py +++ b/nanobot/agent/tools/filesystem.py @@ -5,7 +5,8 @@ import mimetypes from pathlib import Path from typing import Any -from nanobot.agent.tools.base import Tool +from nanobot.agent.tools.base import Tool, tool_parameters +from nanobot.agent.tools.schema import BooleanSchema, IntegerSchema, StringSchema, tool_parameters_schema from nanobot.utils.helpers import build_image_content_blocks, detect_image_mime from nanobot.config.paths import get_media_dir @@ -58,6 +59,23 @@ class _FsTool(Tool): # read_file # --------------------------------------------------------------------------- + +@tool_parameters( + tool_parameters_schema( + path=StringSchema("The file path to read"), + offset=IntegerSchema( + 1, + description="Line number to start reading from (1-indexed, default 1)", + minimum=1, + ), + limit=IntegerSchema( + 2000, + description="Maximum number of lines to read (default 2000)", + minimum=1, + ), + required=["path"], + ) +) class ReadFileTool(_FsTool): """Read file contents with optional line-based pagination.""" @@ -79,26 +97,6 @@ class ReadFileTool(_FsTool): def read_only(self) -> bool: return True - @property - def parameters(self) -> dict[str, Any]: - return { - "type": "object", - "properties": { - "path": {"type": "string", "description": "The file path to read"}, - "offset": { - "type": "integer", - "description": "Line number to start reading from (1-indexed, default 1)", - "minimum": 1, - }, - "limit": { - "type": "integer", - "description": "Maximum number of lines to read (default 2000)", - "minimum": 1, - }, - }, - "required": ["path"], - } - async def execute(self, path: str | None = None, offset: int = 1, limit: int | None = None, **kwargs: Any) -> Any: try: if not path: @@ -160,6 +158,14 @@ class ReadFileTool(_FsTool): # write_file # --------------------------------------------------------------------------- + +@tool_parameters( + tool_parameters_schema( + path=StringSchema("The file path to write to"), + content=StringSchema("The content to write"), + required=["path", "content"], + ) +) class WriteFileTool(_FsTool): """Write content to a file.""" @@ -171,17 +177,6 @@ class WriteFileTool(_FsTool): def description(self) -> str: return "Write content to a file at the given path. Creates parent directories if needed." - @property - def parameters(self) -> dict[str, Any]: - return { - "type": "object", - "properties": { - "path": {"type": "string", "description": "The file path to write to"}, - "content": {"type": "string", "description": "The content to write"}, - }, - "required": ["path", "content"], - } - async def execute(self, path: str | None = None, content: str | None = None, **kwargs: Any) -> str: try: if not path: @@ -228,6 +223,15 @@ def _find_match(content: str, old_text: str) -> tuple[str | None, int]: return None, 0 +@tool_parameters( + tool_parameters_schema( + path=StringSchema("The file path to edit"), + old_text=StringSchema("The text to find and replace"), + new_text=StringSchema("The text to replace with"), + replace_all=BooleanSchema(description="Replace all occurrences (default false)"), + required=["path", "old_text", "new_text"], + ) +) class EditFileTool(_FsTool): """Edit a file by replacing text with fallback matching.""" @@ -243,22 +247,6 @@ class EditFileTool(_FsTool): "Set replace_all=true to replace every occurrence." ) - @property - def parameters(self) -> dict[str, Any]: - return { - "type": "object", - "properties": { - "path": {"type": "string", "description": "The file path to edit"}, - "old_text": {"type": "string", "description": "The text to find and replace"}, - "new_text": {"type": "string", "description": "The text to replace with"}, - "replace_all": { - "type": "boolean", - "description": "Replace all occurrences (default false)", - }, - }, - "required": ["path", "old_text", "new_text"], - } - async def execute( self, path: str | None = None, old_text: str | None = None, new_text: str | None = None, @@ -328,6 +316,18 @@ class EditFileTool(_FsTool): # list_dir # --------------------------------------------------------------------------- +@tool_parameters( + tool_parameters_schema( + path=StringSchema("The directory path to list"), + recursive=BooleanSchema(description="Recursively list all files (default false)"), + max_entries=IntegerSchema( + 200, + description="Maximum entries to return (default 200)", + minimum=1, + ), + required=["path"], + ) +) class ListDirTool(_FsTool): """List directory contents with optional recursion.""" @@ -354,25 +354,6 @@ class ListDirTool(_FsTool): def read_only(self) -> bool: return True - @property - def parameters(self) -> dict[str, Any]: - return { - "type": "object", - "properties": { - "path": {"type": "string", "description": "The directory path to list"}, - "recursive": { - "type": "boolean", - "description": "Recursively list all files (default false)", - }, - "max_entries": { - "type": "integer", - "description": "Maximum entries to return (default 200)", - "minimum": 1, - }, - }, - "required": ["path"], - } - async def execute( self, path: str | None = None, recursive: bool = False, max_entries: int | None = None, **kwargs: Any, diff --git a/nanobot/agent/tools/message.py b/nanobot/agent/tools/message.py index 520020735..524cadcf5 100644 --- a/nanobot/agent/tools/message.py +++ b/nanobot/agent/tools/message.py @@ -2,10 +2,23 @@ from typing import Any, Awaitable, Callable -from nanobot.agent.tools.base import Tool +from nanobot.agent.tools.base import Tool, tool_parameters +from nanobot.agent.tools.schema import ArraySchema, StringSchema, tool_parameters_schema from nanobot.bus.events import OutboundMessage +@tool_parameters( + tool_parameters_schema( + content=StringSchema("The message content to send"), + channel=StringSchema("Optional: target channel (telegram, discord, etc.)"), + chat_id=StringSchema("Optional: target chat/user ID"), + media=ArraySchema( + StringSchema(""), + description="Optional: list of file paths to attach (images, audio, documents)", + ), + required=["content"], + ) +) class MessageTool(Tool): """Tool to send messages to users on chat channels.""" @@ -49,32 +62,6 @@ class MessageTool(Tool): "Do NOT use read_file to send files — that only reads content for your own analysis." ) - @property - def parameters(self) -> dict[str, Any]: - return { - "type": "object", - "properties": { - "content": { - "type": "string", - "description": "The message content to send" - }, - "channel": { - "type": "string", - "description": "Optional: target channel (telegram, discord, etc.)" - }, - "chat_id": { - "type": "string", - "description": "Optional: target chat/user ID" - }, - "media": { - "type": "array", - "items": {"type": "string"}, - "description": "Optional: list of file paths to attach (images, audio, documents)" - } - }, - "required": ["content"] - } - async def execute( self, content: str, diff --git a/nanobot/agent/tools/schema.py b/nanobot/agent/tools/schema.py new file mode 100644 index 000000000..2b7016d74 --- /dev/null +++ b/nanobot/agent/tools/schema.py @@ -0,0 +1,232 @@ +"""JSON Schema fragment types: all subclass :class:`~nanobot.agent.tools.base.Schema` for descriptions and constraints on tool parameters. + +- ``to_json_schema()``: returns a dict compatible with :meth:`~nanobot.agent.tools.base.Schema.validate_json_schema_value` / + :class:`~nanobot.agent.tools.base.Tool`. +- ``validate_value(value, path)``: validates a single value against this schema; returns a list of error messages (empty means valid). + +Shared validation and fragment normalization are on the class methods of :class:`~nanobot.agent.tools.base.Schema`. + +Note: Python does not allow subclassing ``bool``, so booleans use :class:`BooleanSchema`. +""" + +from __future__ import annotations + +from collections.abc import Mapping +from typing import Any + +from nanobot.agent.tools.base import Schema + + +class StringSchema(Schema): + """String parameter: ``description`` documents the field; optional length bounds and enum.""" + + def __init__( + self, + description: str = "", + *, + min_length: int | None = None, + max_length: int | None = None, + enum: tuple[Any, ...] | list[Any] | None = None, + nullable: bool = False, + ) -> None: + self._description = description + self._min_length = min_length + self._max_length = max_length + self._enum = tuple(enum) if enum is not None else None + self._nullable = nullable + + def to_json_schema(self) -> dict[str, Any]: + t: Any = "string" + if self._nullable: + t = ["string", "null"] + d: dict[str, Any] = {"type": t} + if self._description: + d["description"] = self._description + if self._min_length is not None: + d["minLength"] = self._min_length + if self._max_length is not None: + d["maxLength"] = self._max_length + if self._enum is not None: + d["enum"] = list(self._enum) + return d + + +class IntegerSchema(Schema): + """Integer parameter: optional placeholder int (legacy ctor signature), description, and bounds.""" + + def __init__( + self, + value: int = 0, + *, + description: str = "", + minimum: int | None = None, + maximum: int | None = None, + enum: tuple[int, ...] | list[int] | None = None, + nullable: bool = False, + ) -> None: + self._value = value + self._description = description + self._minimum = minimum + self._maximum = maximum + self._enum = tuple(enum) if enum is not None else None + self._nullable = nullable + + def to_json_schema(self) -> dict[str, Any]: + t: Any = "integer" + if self._nullable: + t = ["integer", "null"] + d: dict[str, Any] = {"type": t} + if self._description: + d["description"] = self._description + if self._minimum is not None: + d["minimum"] = self._minimum + if self._maximum is not None: + d["maximum"] = self._maximum + if self._enum is not None: + d["enum"] = list(self._enum) + return d + + +class NumberSchema(Schema): + """Numeric parameter (JSON number): description and optional bounds.""" + + def __init__( + self, + value: float = 0.0, + *, + description: str = "", + minimum: float | None = None, + maximum: float | None = None, + enum: tuple[float, ...] | list[float] | None = None, + nullable: bool = False, + ) -> None: + self._value = value + self._description = description + self._minimum = minimum + self._maximum = maximum + self._enum = tuple(enum) if enum is not None else None + self._nullable = nullable + + def to_json_schema(self) -> dict[str, Any]: + t: Any = "number" + if self._nullable: + t = ["number", "null"] + d: dict[str, Any] = {"type": t} + if self._description: + d["description"] = self._description + if self._minimum is not None: + d["minimum"] = self._minimum + if self._maximum is not None: + d["maximum"] = self._maximum + if self._enum is not None: + d["enum"] = list(self._enum) + return d + + +class BooleanSchema(Schema): + """Boolean parameter (standalone class because Python forbids subclassing ``bool``).""" + + def __init__( + self, + *, + description: str = "", + default: bool | None = None, + nullable: bool = False, + ) -> None: + self._description = description + self._default = default + self._nullable = nullable + + def to_json_schema(self) -> dict[str, Any]: + t: Any = "boolean" + if self._nullable: + t = ["boolean", "null"] + d: dict[str, Any] = {"type": t} + if self._description: + d["description"] = self._description + if self._default is not None: + d["default"] = self._default + return d + + +class ArraySchema(Schema): + """Array parameter: element schema is given by ``items``.""" + + def __init__( + self, + items: Any | None = None, + *, + description: str = "", + min_items: int | None = None, + max_items: int | None = None, + nullable: bool = False, + ) -> None: + self._items_schema: Any = items if items is not None else StringSchema("") + self._description = description + self._min_items = min_items + self._max_items = max_items + self._nullable = nullable + + def to_json_schema(self) -> dict[str, Any]: + t: Any = "array" + if self._nullable: + t = ["array", "null"] + d: dict[str, Any] = { + "type": t, + "items": Schema.fragment(self._items_schema), + } + if self._description: + d["description"] = self._description + if self._min_items is not None: + d["minItems"] = self._min_items + if self._max_items is not None: + d["maxItems"] = self._max_items + return d + + +class ObjectSchema(Schema): + """Object parameter: ``properties`` or keyword args are field names; values are child Schema or JSON Schema dicts.""" + + def __init__( + self, + properties: Mapping[str, Any] | None = None, + *, + required: list[str] | None = None, + description: str = "", + additional_properties: bool | dict[str, Any] | None = None, + nullable: bool = False, + **kwargs: Any, + ) -> None: + self._properties = dict(properties or {}, **kwargs) + self._required = list(required or []) + self._root_description = description + self._additional_properties = additional_properties + self._nullable = nullable + + def to_json_schema(self) -> dict[str, Any]: + t: Any = "object" + if self._nullable: + t = ["object", "null"] + props = {k: Schema.fragment(v) for k, v in self._properties.items()} + out: dict[str, Any] = {"type": t, "properties": props} + if self._required: + out["required"] = self._required + if self._root_description: + out["description"] = self._root_description + if self._additional_properties is not None: + out["additionalProperties"] = self._additional_properties + return out + + +def tool_parameters_schema( + *, + required: list[str] | None = None, + description: str = "", + **properties: Any, +) -> dict[str, Any]: + """Build root tool parameters ``{"type": "object", "properties": ...}`` for :meth:`Tool.parameters`.""" + return ObjectSchema( + required=required, + description=description, + **properties, + ).to_json_schema() diff --git a/nanobot/agent/tools/shell.py b/nanobot/agent/tools/shell.py index c987a5f99..c8876827c 100644 --- a/nanobot/agent/tools/shell.py +++ b/nanobot/agent/tools/shell.py @@ -9,10 +9,27 @@ from typing import Any from loguru import logger -from nanobot.agent.tools.base import Tool +from nanobot.agent.tools.base import Tool, tool_parameters +from nanobot.agent.tools.schema import IntegerSchema, StringSchema, tool_parameters_schema from nanobot.config.paths import get_media_dir +@tool_parameters( + tool_parameters_schema( + command=StringSchema("The shell command to execute"), + working_dir=StringSchema("Optional working directory for the command"), + timeout=IntegerSchema( + 60, + description=( + "Timeout in seconds. Increase for long-running commands " + "like compilation or installation (default 60, max 600)." + ), + minimum=1, + maximum=600, + ), + required=["command"], + ) +) class ExecTool(Tool): """Tool to execute shell commands.""" @@ -57,32 +74,6 @@ class ExecTool(Tool): def exclusive(self) -> bool: return True - @property - def parameters(self) -> dict[str, Any]: - return { - "type": "object", - "properties": { - "command": { - "type": "string", - "description": "The shell command to execute", - }, - "working_dir": { - "type": "string", - "description": "Optional working directory for the command", - }, - "timeout": { - "type": "integer", - "description": ( - "Timeout in seconds. Increase for long-running commands " - "like compilation or installation (default 60, max 600)." - ), - "minimum": 1, - "maximum": 600, - }, - }, - "required": ["command"], - } - async def execute( self, command: str, working_dir: str | None = None, timeout: int | None = None, **kwargs: Any, diff --git a/nanobot/agent/tools/spawn.py b/nanobot/agent/tools/spawn.py index 2050eed22..86319e991 100644 --- a/nanobot/agent/tools/spawn.py +++ b/nanobot/agent/tools/spawn.py @@ -2,12 +2,20 @@ from typing import TYPE_CHECKING, Any -from nanobot.agent.tools.base import Tool +from nanobot.agent.tools.base import Tool, tool_parameters +from nanobot.agent.tools.schema import StringSchema, tool_parameters_schema if TYPE_CHECKING: from nanobot.agent.subagent import SubagentManager +@tool_parameters( + tool_parameters_schema( + task=StringSchema("The task for the subagent to complete"), + label=StringSchema("Optional short label for the task (for display)"), + required=["task"], + ) +) class SpawnTool(Tool): """Tool to spawn a subagent for background task execution.""" @@ -37,23 +45,6 @@ class SpawnTool(Tool): "and use a dedicated subdirectory when helpful." ) - @property - def parameters(self) -> dict[str, Any]: - return { - "type": "object", - "properties": { - "task": { - "type": "string", - "description": "The task for the subagent to complete", - }, - "label": { - "type": "string", - "description": "Optional short label for the task (for display)", - }, - }, - "required": ["task"], - } - async def execute(self, task: str, label: str | None = None, **kwargs: Any) -> str: """Spawn a subagent to execute the given task.""" return await self._manager.spawn( diff --git a/nanobot/agent/tools/web.py b/nanobot/agent/tools/web.py index 1c0fde822..9ac923050 100644 --- a/nanobot/agent/tools/web.py +++ b/nanobot/agent/tools/web.py @@ -13,7 +13,8 @@ from urllib.parse import urlparse import httpx from loguru import logger -from nanobot.agent.tools.base import Tool +from nanobot.agent.tools.base import Tool, tool_parameters +from nanobot.agent.tools.schema import IntegerSchema, StringSchema, tool_parameters_schema from nanobot.utils.helpers import build_image_content_blocks if TYPE_CHECKING: @@ -72,19 +73,18 @@ def _format_results(query: str, items: list[dict[str, Any]], n: int) -> str: return "\n".join(lines) +@tool_parameters( + tool_parameters_schema( + query=StringSchema("Search query"), + count=IntegerSchema(1, description="Results (1-10)", minimum=1, maximum=10), + required=["query"], + ) +) class WebSearchTool(Tool): """Search the web using configured provider.""" name = "web_search" description = "Search the web. Returns titles, URLs, and snippets." - parameters = { - "type": "object", - "properties": { - "query": {"type": "string", "description": "Search query"}, - "count": {"type": "integer", "description": "Results (1-10)", "minimum": 1, "maximum": 10}, - }, - "required": ["query"], - } def __init__(self, config: WebSearchConfig | None = None, proxy: str | None = None): from nanobot.config.schema import WebSearchConfig @@ -219,20 +219,23 @@ class WebSearchTool(Tool): return f"Error: DuckDuckGo search failed ({e})" +@tool_parameters( + tool_parameters_schema( + url=StringSchema("URL to fetch"), + extractMode={ + "type": "string", + "enum": ["markdown", "text"], + "default": "markdown", + }, + maxChars=IntegerSchema(0, minimum=100), + required=["url"], + ) +) class WebFetchTool(Tool): """Fetch and extract content from a URL.""" name = "web_fetch" description = "Fetch URL and extract readable content (HTML → markdown/text)." - parameters = { - "type": "object", - "properties": { - "url": {"type": "string", "description": "URL to fetch"}, - "extractMode": {"type": "string", "enum": ["markdown", "text"], "default": "markdown"}, - "maxChars": {"type": "integer", "minimum": 100}, - }, - "required": ["url"], - } def __init__(self, max_chars: int = 50000, proxy: str | None = None): self.max_chars = max_chars diff --git a/tests/tools/test_tool_validation.py b/tests/tools/test_tool_validation.py index 0fd15e383..b1d56a439 100644 --- a/tests/tools/test_tool_validation.py +++ b/tests/tools/test_tool_validation.py @@ -1,5 +1,13 @@ from typing import Any +from nanobot.agent.tools import ( + ArraySchema, + IntegerSchema, + ObjectSchema, + Schema, + StringSchema, + tool_parameters_schema, +) from nanobot.agent.tools.base import Tool from nanobot.agent.tools.registry import ToolRegistry from nanobot.agent.tools.shell import ExecTool @@ -41,6 +49,58 @@ class SampleTool(Tool): return "ok" +def test_schema_validate_value_matches_tool_validate_params() -> None: + """ObjectSchema.validate_value 与 validate_json_schema_value、Tool.validate_params 一致。""" + root = tool_parameters_schema( + query=StringSchema(min_length=2), + count=IntegerSchema(2, minimum=1, maximum=10), + required=["query", "count"], + ) + obj = ObjectSchema( + query=StringSchema(min_length=2), + count=IntegerSchema(2, minimum=1, maximum=10), + required=["query", "count"], + ) + params = {"query": "h", "count": 2} + + class _Mini(Tool): + @property + def name(self) -> str: + return "m" + + @property + def description(self) -> str: + return "" + + @property + def parameters(self) -> dict[str, Any]: + return root + + async def execute(self, **kwargs: Any) -> str: + return "" + + expected = _Mini().validate_params(params) + assert Schema.validate_json_schema_value(params, root, "") == expected + assert obj.validate_value(params, "") == expected + assert IntegerSchema(0, minimum=1).validate_value(0, "n") == ["n must be >= 1"] + + +def test_schema_classes_equivalent_to_sample_tool_parameters() -> None: + """Schema 类生成的 JSON Schema 应与手写 dict 一致，便于校验行为一致。""" + built = tool_parameters_schema( + query=StringSchema(min_length=2), + count=IntegerSchema(2, minimum=1, maximum=10), + mode=StringSchema("", enum=["fast", "full"]), + meta=ObjectSchema( + tag=StringSchema(""), + flags=ArraySchema(StringSchema("")), + required=["tag"], + ), + required=["query", "count"], + ) + assert built == SampleTool().parameters + + def test_validate_params_missing_required() -> None: tool = SampleTool() errors = tool.validate_params({"query": "hi"})