File size: 1,734 Bytes
f0f6e5c
 
 
 
406f126
f0f6e5c
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
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 function is not callable and isn't decorated using attach_param_schema
            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.")
            # Handle multi-line docstirngs
            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):
            # Throw an error if there's a mismatch between the function parameters and pydantic model's fields.
            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