base_tool.py

import json
import inspect
from abc import ABC, abstractmethod
from typing import Dict, List, Optional, Union

# Global tool registry
TOOL_REGISTRY = {}

def register_tool(name: str):
    """Decorator to register a tool in the global registry."""
    def decorator(cls):
        cls.name = name
        TOOL_REGISTRY[name] = cls
        return cls
    return decorator

class BaseTool(ABC):
    """Base class for all system tools matching Qwen-Agent API protocol."""
    name: str = ''
    description: str = ''
    _static_parameters: dict = {}

    def __init__(self, cfg: Optional[dict] = None):
        self.cfg = cfg or {}

    @property
    def parameters(self) -> dict:
        """Dynamically autodetect and build Qwen-Agent schema from type annotations if not explicitly set."""
        # 1. If explicit static parameters are defined on the class, use them
        static_p = getattr(self, '_static_parameters', None)
        if not static_p:
            static_p = self.__class__.__dict__.get('parameters')
            
        if static_p:
            return static_p
            
        # 2. Try autodetecting from 'execute' or 'run' methods using reflection and type annotations
        target_method = None
        for m_name in ['execute', 'run']:
            if hasattr(self, m_name):
                target_method = getattr(self, m_name)
                break
                
        if not target_method:
            return {"type": "object", "properties": {}, "required": []}
            
        try:
            sig = inspect.signature(target_method)
            properties = {}
            required = []
            
            type_map = {
                str: "string",
                int: "integer",
                float: "number",
                bool: "boolean",
                list: "array",
                dict: "object"
            }
            
            doc = target_method.__doc__ or ""
            for param_name, param in sig.parameters.items():
                if param_name in ['self', 'args', 'kwargs']:
                    continue
                
                # Map type annotations
                anno = param.annotation
                p_type = "string" # fallback
                if anno in type_map:
                    p_type = type_map[anno]
                elif hasattr(anno, "__origin__") and anno.__origin__ in [list, List]:
                    p_type = "array"
                elif hasattr(anno, "__origin__") and anno.__origin__ in [dict, Dict]:
                    p_type = "object"
                    
                # Extract parameter description from docstring if available
                desc = f"Parameter {param_name}"
                for line in doc.split("\n"):
                    if param_name in line and ":" in line:
                        desc = line.split(":", 1)[1].strip()
                        break
                        
                properties[param_name] = {
                    "type": p_type,
                    "description": desc
                }
                
                if param.default == inspect.Parameter.empty:
                    required.append(param_name)
                    
            return {
                "type": "object",
                "properties": properties,
                "required": required
            }
        except Exception:
            return {"type": "object", "properties": {}, "required": []}

    @abstractmethod
    def call(self, params: Union[str, dict], **kwargs) -> str:
        """Executes the tool with the given parameters and returns a string result."""
        raise NotImplementedError

    @property
    def function(self) -> dict:
        """Returns the schema description of the tool compatible with OpenAI/Qwen."""
        return {
            'name': self.name,
            'description': self.description,
            'parameters': self.parameters,
        }