|
import inspect |
|
from functools import cached_property, wraps |
|
from typing import Any, Callable, Optional |
|
|
|
from pydantic import BaseModel |
|
|
|
|
|
class Tool: |
|
async def __aenter__(self): |
|
pass |
|
|
|
async def __aexit__(self, exc_type, exc_val, exc_tb): |
|
pass |
|
|
|
@cached_property |
|
def schema(self) -> list[dict[str, Any]]: |
|
schema = [] |
|
for name, method in self.__class__.__dict__.items(): |
|
|
|
if not isinstance(method, Callable) or not hasattr(method, "param_model"): |
|
continue |
|
|
|
docstring = inspect.getdoc(method) |
|
if not docstring: |
|
raise ValueError(f"The tool function '{name}' is missing a docstring.") |
|
|
|
description = " ".join(line.strip() for line in docstring.split("\n")) |
|
|
|
tool_json = { |
|
"name": name, |
|
"description": description, |
|
"parameters": method.param_model.model_json_schema(), |
|
} |
|
schema.append(tool_json) |
|
return schema |
|
|
|
|
|
def attach_param_schema(param_model: type[BaseModel]): |
|
def decorator(func: Callable) -> Callable: |
|
@wraps(func) |
|
def wrapper(self, **kwargs): |
|
|
|
validated_params = param_model(**kwargs) |
|
return func(self, **validated_params.model_dump()) |
|
|
|
wrapper.param_model = param_model |
|
return wrapper |
|
|
|
return decorator |
|
|
|
|
|
class ToolExecutionResponse(BaseModel): |
|
content: Optional[str] = None |
|
id: Optional[str] = None |
|
|