Spaces:
Paused
Paused
File size: 3,420 Bytes
0d3f7cc | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 | """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
@property
def name(self) -> str:
"""Tool name."""
return self.schema.name
@property
def schema(self) -> ToolSchema:
"""Get tool schema."""
if self._schema is None:
self._schema = self._define_schema()
return self._schema
@abstractmethod
def _define_schema(self) -> ToolSchema:
"""Define the tool's schema."""
@abstractmethod
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,
},
}
|