Skip to content

activity_mixin

ActivityMixin

Provides Tool Activity management functionality to Tools.

Attributes:

Name Type Description
allowlist Optional[list[str]]

List of Tool Activities to include in the Tool schema.
denylist Optional[list[str]]

List of Tool Activities to remove from the Tool schema.
extra_schema_properties Optional[dict[str, dict]]

Mapping of Activity name and extra properties to include in the activity's schema.
Source code in griptape/mixins/activity_mixin.py 
 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
119
120
121
122
123
124
125
126
127
128
129
130
131
@define(slots=False)
class ActivityMixin:
    """Provides Tool Activity management functionality to Tools.

    Attributes:
        allowlist: List of Tool Activities to include in the Tool schema.
        denylist: List of Tool Activities to remove from the Tool schema.
        extra_schema_properties: Mapping of Activity name and extra properties to include in the activity's schema.
    """

    allowlist: Optional[list[str]] = field(default=None, kw_only=True)
    denylist: Optional[list[str]] = field(default=None, kw_only=True)
    extra_schema_properties: Optional[dict[str, dict]] = field(default=None, kw_only=True)

    @allowlist.validator  # pyright: ignore[reportAttributeAccessIssue, reportOptionalMemberAccess]
    def validate_allowlist(self, _: Attribute, allowlist: Optional[list[str]]) -> None:
        if allowlist is None:
            return

        for activity_name in allowlist:
            self._validate_tool_activity(activity_name)

    @denylist.validator  # pyright: ignore[reportAttributeAccessIssue, reportOptionalMemberAccess]
    def validate_denylist(self, _: Attribute, denylist: Optional[list[str]]) -> None:
        if denylist is None:
            return

        for activity_name in denylist:
            self._validate_tool_activity(activity_name)

    def enable_activities(self) -> None:
        self.allowlist = None
        self.denylist = None

    def disable_activities(self) -> None:
        self.allowlist = []
        self.denylist = None

    # This method has to remain a method and can't be decorated with @property because
    # of the max depth recursion issue in `inspect.getmembers`.
    def activities(self) -> list[Callable]:
        methods = []

        for name, method in inspect.getmembers(self, predicate=inspect.ismethod):
            allowlist_condition = self.allowlist is None or name in self.allowlist
            denylist_condition = self.denylist is None or name not in self.denylist

            if getattr(method, "is_activity", False) and allowlist_condition and denylist_condition:
                methods.append(method)

        return methods

    def find_activity(self, name: str) -> Optional[Callable]:
        for activity in self.activities():
            if getattr(activity, "is_activity", False) and getattr(activity, "name") == name:
                return activity

        return None

    def activity_name(self, activity: Callable) -> str:
        if activity is None or not getattr(activity, "is_activity", False):
            raise Exception("This method is not an activity.")
        return getattr(activity, "name")

    def activity_description(self, activity: Callable) -> str:
        if activity is None or not getattr(activity, "is_activity", False):
            raise Exception("This method is not an activity.")
        return Template(getattr(activity, "config")["description"]).render({"_self": self})

    def activity_schema(self, activity: Callable) -> Optional[Union[Schema, type[BaseModel]]]:
        if activity is None or not getattr(activity, "is_activity", False):
            raise Exception("This method is not an activity.")
        if getattr(activity, "config")["schema"] is not None:
            config_schema = getattr(activity, "config")["schema"]
            if isinstance(config_schema, Schema):
                # Need to deepcopy to avoid modifying the original schema
                config_schema = deepcopy(getattr(activity, "config")["schema"])
            elif isinstance(config_schema, Callable) and not isinstance(config_schema, type):
                config_schema = config_schema(self)
            activity_name = self.activity_name(activity)

            if isinstance(config_schema, Schema):
                if self.extra_schema_properties is not None and activity_name in self.extra_schema_properties:
                    config_schema.schema.update(self.extra_schema_properties[activity_name])

                return Schema({"values": config_schema})
            return create_model(config_schema.__name__, values=(config_schema, Field(config_schema)))
        return None

    def to_activity_json_schema(self, activity: Callable, schema_id: str) -> dict:
        schema = self.activity_schema(activity)
        if schema is None:
            schema = Schema({})

        if isinstance(schema, Schema):
            json_schema = schema.json_schema(schema_id)
        else:
            json_schema = build_strict_schema(schema.model_json_schema(), schema_id)

        return json_schema

    def validate_activity_schema(self, activity_schema: Union[Schema, type[BaseModel]], params: dict) -> None:
        try:
            if isinstance(activity_schema, Schema):
                activity_schema.validate(params)
            else:
                activity_schema.model_validate(params)
        except (SchemaError, ValidationError) as e:
            raise ValueError(e) from e

    def _validate_tool_activity(self, activity_name: str) -> None:
        tool = self.__class__

        activity = getattr(tool, activity_name, None)

        if not activity or not getattr(activity, "is_activity", False):
            raise ValueError(f"activity {activity_name} is not a valid activity for {tool}")

allowlist = field(default=None, kw_only=True) class-attribute instance-attribute

denylist = field(default=None, kw_only=True) class-attribute instance-attribute

extra_schema_properties = field(default=None, kw_only=True) class-attribute instance-attribute

_validate_tool_activity(activity_name)

Source code in griptape/mixins/activity_mixin.py 
125
126
127
128
129
130
131
def _validate_tool_activity(self, activity_name: str) -> None:
    tool = self.__class__

    activity = getattr(tool, activity_name, None)

    if not activity or not getattr(activity, "is_activity", False):
        raise ValueError(f"activity {activity_name} is not a valid activity for {tool}")

activities()

Source code in griptape/mixins/activity_mixin.py 
55
56
57
58
59
60
61
62
63
64
65
def activities(self) -> list[Callable]:
    methods = []

    for name, method in inspect.getmembers(self, predicate=inspect.ismethod):
        allowlist_condition = self.allowlist is None or name in self.allowlist
        denylist_condition = self.denylist is None or name not in self.denylist

        if getattr(method, "is_activity", False) and allowlist_condition and denylist_condition:
            methods.append(method)

    return methods

activity_description(activity)

Source code in griptape/mixins/activity_mixin.py 
79
80
81
82
def activity_description(self, activity: Callable) -> str:
    if activity is None or not getattr(activity, "is_activity", False):
        raise Exception("This method is not an activity.")
    return Template(getattr(activity, "config")["description"]).render({"_self": self})

activity_name(activity)

Source code in griptape/mixins/activity_mixin.py 
74
75
76
77
def activity_name(self, activity: Callable) -> str:
    if activity is None or not getattr(activity, "is_activity", False):
        raise Exception("This method is not an activity.")
    return getattr(activity, "name")

activity_schema(activity)

Source code in griptape/mixins/activity_mixin.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
def activity_schema(self, activity: Callable) -> Optional[Union[Schema, type[BaseModel]]]:
    if activity is None or not getattr(activity, "is_activity", False):
        raise Exception("This method is not an activity.")
    if getattr(activity, "config")["schema"] is not None:
        config_schema = getattr(activity, "config")["schema"]
        if isinstance(config_schema, Schema):
            # Need to deepcopy to avoid modifying the original schema
            config_schema = deepcopy(getattr(activity, "config")["schema"])
        elif isinstance(config_schema, Callable) and not isinstance(config_schema, type):
            config_schema = config_schema(self)
        activity_name = self.activity_name(activity)

        if isinstance(config_schema, Schema):
            if self.extra_schema_properties is not None and activity_name in self.extra_schema_properties:
                config_schema.schema.update(self.extra_schema_properties[activity_name])

            return Schema({"values": config_schema})
        return create_model(config_schema.__name__, values=(config_schema, Field(config_schema)))
    return None

disable_activities()

Source code in griptape/mixins/activity_mixin.py 
49
50
51
def disable_activities(self) -> None:
    self.allowlist = []
    self.denylist = None

enable_activities()

Source code in griptape/mixins/activity_mixin.py 
45
46
47
def enable_activities(self) -> None:
    self.allowlist = None
    self.denylist = None

find_activity(name)

Source code in griptape/mixins/activity_mixin.py 
67
68
69
70
71
72
def find_activity(self, name: str) -> Optional[Callable]:
    for activity in self.activities():
        if getattr(activity, "is_activity", False) and getattr(activity, "name") == name:
            return activity

    return None

to_activity_json_schema(activity, schema_id)

Source code in griptape/mixins/activity_mixin.py 
104
105
106
107
108
109
110
111
112
113
114
def to_activity_json_schema(self, activity: Callable, schema_id: str) -> dict:
    schema = self.activity_schema(activity)
    if schema is None:
        schema = Schema({})

    if isinstance(schema, Schema):
        json_schema = schema.json_schema(schema_id)
    else:
        json_schema = build_strict_schema(schema.model_json_schema(), schema_id)

    return json_schema

validate_activity_schema(activity_schema, params)

Source code in griptape/mixins/activity_mixin.py 
116
117
118
119
120
121
122
123
def validate_activity_schema(self, activity_schema: Union[Schema, type[BaseModel]], params: dict) -> None:
    try:
        if isinstance(activity_schema, Schema):
            activity_schema.validate(params)
        else:
            activity_schema.model_validate(params)
    except (SchemaError, ValidationError) as e:
        raise ValueError(e) from e

validate_allowlist(_, allowlist)

Source code in griptape/mixins/activity_mixin.py 
29
30
31
32
33
34
35
@allowlist.validator  # pyright: ignore[reportAttributeAccessIssue, reportOptionalMemberAccess]
def validate_allowlist(self, _: Attribute, allowlist: Optional[list[str]]) -> None:
    if allowlist is None:
        return

    for activity_name in allowlist:
        self._validate_tool_activity(activity_name)

validate_denylist(_, denylist)

Source code in griptape/mixins/activity_mixin.py 
37
38
39
40
41
42
43
@denylist.validator  # pyright: ignore[reportAttributeAccessIssue, reportOptionalMemberAccess]
def validate_denylist(self, _: Attribute, denylist: Optional[list[str]]) -> None:
    if denylist is None:
        return

    for activity_name in denylist:
        self._validate_tool_activity(activity_name)