refactor(tools): streamline Tool class and add JSON Schema for parameters

Refactor Tool methods and type handling; introduce JSON Schema support for tool parameters (schema module, validation tests).

Made-with: Cursor

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",
+]

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):
-    """
+    """Agent capability: read files, run commands, etc."""
-    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.
-    """
 
     _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.
+        """Pick first non-null type from JSON Schema unions like ``['string','null']``."""
-
+        return Schema.resolve_json_schema_type(t)
-        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
 
     @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:
-        """
+        """Run the tool; returns a string or list of content blocks."""
-        Execute the tool with given parameters.
+        ...
 
-        Args:
+    def _cast_object(self, obj: Any, schema: dict[str, Any]) -> dict[str, Any]:
-            **kwargs: Tool-specific parameters.
+        if not isinstance(obj, dict):
-
+            return obj
-        Returns:
+        props = schema.get("properties", {})
-            Result of the tool execution (string or list of content blocks).
+        return {k: self._cast_value(v, props[k]) if k in props else v for k, v in obj.items()}
-        """
-        pass
 
     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."""
+        t = self._resolve_type(schema.get("type"))
-        target_type = 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"):
+        if t in self._TYPE_MAP and t not in ("boolean", "integer", "array", "object"):
-            expected = self._TYPE_MAP[target_type]
+            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):
+        if t == "string":
-            try:
-                return float(val)
-            except ValueError:
-                return val
-
-        if target_type == "string":
             return val if val is None else str(val)
 
-        if target_type == "boolean" and isinstance(val, str):
+        if t == "boolean" and isinstance(val, str):
-            val_lower = val.lower()
+            low = val.lower()
-            if val_lower in ("true", "1", "yes"):
+            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):
+        if t == "array" and isinstance(val, list):
-            item_schema = schema.get("items")
+            items = schema.get("items")
-            return [self._cast_value(item, item_schema) for item in val] if item_schema else val
+            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"}, "")
+        return Schema.validate_json_schema_value(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
 
     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

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,

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,

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,

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()

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,

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(

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

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"})