File size: 3,590 Bytes
068bc7f 13d8bea 068bc7f 13d8bea b64b6b0 13d8bea 068bc7f b64b6b0 13d8bea b64b6b0 068bc7f | 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 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 | """Base tool class for Stack 2.9 tools."""
from __future__ import annotations
import asyncio
import inspect
import time
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Any, Callable, Generic, TypeVar
TInput = TypeVar("TInput")
TOutput = TypeVar("TOutput")
@dataclass
class ToolParam:
"""Definition of a tool parameter."""
name: str
description: str
type: str = "string"
required: bool = True
default: Any = None
@dataclass
class ToolResult:
"""Result returned by a tool execution."""
success: bool = True
data: Any = None
error: str | None = None
duration_seconds: float = 0.0
class BaseTool(ABC, Generic[TInput, TOutput]):
"""Abstract base class for all Stack 2.9 tools.
Subclasses must implement:
- name: str — unique identifier
- description: str — human-readable description
- input_schema: dict — JSON schema for parameters
- execute(input: TInput) -> ToolResult[TOutput]
Optional overrides:
- validate_input(input: dict) -> tuple[bool, str | None]
- is_enabled() -> bool
"""
name: str = ""
description: str = ""
search_hint: str = ""
max_result_size_chars: int = 100_000
@property
def input_schema(self) -> dict[str, Any]:
"""JSON schema for tool input parameters."""
return {}
@property
def output_schema(self) -> dict[str, Any]:
"""JSON schema for tool output."""
return {}
def is_enabled(self) -> bool:
"""Whether the tool is currently available."""
return True
def validate_input(self, input_data: dict[str, Any]) -> tuple[bool, str | None]:
"""Validate input before execution. Returns (valid, error_message)."""
return True, None
@abstractmethod
def execute(self, input_data: TInput) -> ToolResult[TOutput]:
"""Execute the tool with the given input. Must be implemented by subclasses."""
...
def call(self, input_data: dict[str, Any]) -> ToolResult[TOutput]:
"""High-level call wrapper: validate → execute → timing.
Handles both sync and async execute methods, and both
execute(input_data: dict) and execute(path: str, ...) signatures.
"""
valid, error = self.validate_input(input_data)
if not valid:
return ToolResult(success=False, error=error or "Validation failed")
start = time.perf_counter()
try:
# Determine if execute takes a dict or named parameters
sig = inspect.signature(self.execute)
params = list(sig.parameters.keys())
# If first param is 'input_data' (and only one param), pass dict directly
# Otherwise unpack as kwargs
if params == ['input_data']:
result = self.execute(input_data)
else:
result = self.execute(**input_data)
# Handle async execute methods
if inspect.iscoroutine(result):
result = asyncio.run(result)
result.duration_seconds = time.perf_counter() - start
return result
except Exception as exc:
return ToolResult(
success=False,
error=str(exc),
duration_seconds=time.perf_counter() - start,
)
def map_result_to_message(self, result: TOutput, tool_use_id: str | None = None) -> str:
"""Format a successful result for display."""
return str(result)
|