Spaces:
Paused
Paused
| """Base tool abstraction for all tools in the Hermes platform.""" | |
| from __future__ import annotations | |
| import logging | |
| from abc import ABC, abstractmethod | |
| from typing import Any | |
| from pydantic import BaseModel, Field | |
| logger = logging.getLogger(__name__) | |
| class ToolSchema(BaseModel): | |
| """Schema definition for a tool.""" | |
| name: str | |
| description: str | |
| parameters: dict[str, Any] = Field(default_factory=dict) | |
| required: list[str] = Field(default_factory=list) | |
| category: str = "general" | |
| tags: list[str] = Field(default_factory=list) | |
| class BaseTool(ABC): | |
| """Abstract base class for all tools.""" | |
| def __init__(self) -> None: | |
| self._schema: ToolSchema | None = None | |
| def name(self) -> str: | |
| """Tool name.""" | |
| return self.schema.name | |
| def schema(self) -> ToolSchema: | |
| """Get tool schema.""" | |
| if self._schema is None: | |
| self._schema = self._define_schema() | |
| return self._schema | |
| def _define_schema(self) -> ToolSchema: | |
| """Define the tool's schema.""" | |
| async def execute(self, **kwargs: Any) -> Any: | |
| """Execute the tool with given arguments.""" | |
| def validate_arguments(self, arguments: dict[str, Any]) -> dict[str, Any]: | |
| """Validate tool arguments against schema.""" | |
| schema = self.schema | |
| validated = {} | |
| for param_name, param_schema in schema.parameters.items(): | |
| if param_name in arguments: | |
| value = arguments[param_name] | |
| param_type = param_schema.get("type", "string") | |
| if param_type == "string" and not isinstance(value, str): | |
| value = str(value) | |
| elif param_type == "integer" and not isinstance(value, int): | |
| value = int(value) | |
| elif param_type == "number" and not isinstance(value, (int, float)): | |
| value = float(value) | |
| elif param_type == "boolean" and not isinstance(value, bool): | |
| value = bool(value) | |
| validated[param_name] = value | |
| for required_param in schema.required: | |
| if required_param not in validated and required_param not in arguments: | |
| raise ValueError(f"Missing required parameter: {required_param}") | |
| return validated | |
| def to_mcp_schema(self) -> dict[str, Any]: | |
| """Convert to MCP-compatible schema.""" | |
| schema = self.schema | |
| properties = {} | |
| for param_name, param_schema in schema.parameters.items(): | |
| if isinstance(param_schema, dict): | |
| properties[param_name] = { | |
| "type": param_schema.get("type", "string"), | |
| "description": param_schema.get("description", ""), | |
| } | |
| else: | |
| properties[param_name] = { | |
| "type": str(param_schema) if param_schema else "string", | |
| "description": "", | |
| } | |
| return { | |
| "name": schema.name, | |
| "description": schema.description, | |
| "inputSchema": { | |
| "type": "object", | |
| "properties": properties, | |
| "required": schema.required, | |
| }, | |
| } | |