Skip to content

Events

__all__ = ['BaseEvent', 'BaseTaskEvent', 'BaseActionSubtaskEvent', 'StartTaskEvent', 'FinishTaskEvent', 'StartActionSubtaskEvent', 'FinishActionSubtaskEvent', 'StartPromptEvent', 'FinishPromptEvent', 'StartStructureRunEvent', 'FinishStructureRunEvent', 'CompletionChunkEvent', 'EventListener'] module-attribute

BaseActionSubtaskEvent

Bases: BaseTaskEvent, ABC

Source code in griptape/griptape/events/base_action_subtask_event.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
@define
class BaseActionSubtaskEvent(BaseTaskEvent, ABC):
    subtask_parent_task_id: Optional[str] = field(kw_only=True)
    subtask_thought: Optional[str] = field(kw_only=True)
    subtask_action_name: Optional[str] = field(kw_only=True)
    subtask_action_path: Optional[str] = field(kw_only=True)
    subtask_action_input: Optional[dict] = field(kw_only=True)

    @classmethod
    def from_task(cls, task: ActionSubtask) -> BaseActionSubtaskEvent:
        return cls(
            task_id=task.id,
            task_parent_ids=task.parent_ids,
            task_child_ids=task.child_ids,
            task_input=task.input,
            task_output=task.output,
            subtask_parent_task_id=task.parent_task_id,
            subtask_thought=task.thought,
            subtask_action_name=task.action_name,
            subtask_action_path=task.action_path,
            subtask_action_input=task.action_input,
        )

subtask_action_input: Optional[dict] = field(kw_only=True) class-attribute instance-attribute

subtask_action_name: Optional[str] = field(kw_only=True) class-attribute instance-attribute

subtask_action_path: Optional[str] = field(kw_only=True) class-attribute instance-attribute

subtask_parent_task_id: Optional[str] = field(kw_only=True) class-attribute instance-attribute

subtask_thought: Optional[str] = field(kw_only=True) class-attribute instance-attribute

from_task(task) classmethod

Source code in griptape/griptape/events/base_action_subtask_event.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
@classmethod
def from_task(cls, task: ActionSubtask) -> BaseActionSubtaskEvent:
    return cls(
        task_id=task.id,
        task_parent_ids=task.parent_ids,
        task_child_ids=task.child_ids,
        task_input=task.input,
        task_output=task.output,
        subtask_parent_task_id=task.parent_task_id,
        subtask_thought=task.thought,
        subtask_action_name=task.action_name,
        subtask_action_path=task.action_path,
        subtask_action_input=task.action_input,
    )

BaseEvent

Bases: ABC

Source code in griptape/griptape/events/base_event.py 
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
@define
class BaseEvent(ABC):
    timestamp: float = field(default=Factory(lambda: time.time()), kw_only=True)
    type: str = field(default=Factory(lambda self: self.__class__.__name__, takes_self=True), kw_only=True)

    @classmethod
    def from_dict(cls, event_dict: dict) -> BaseEvent:
        from griptape.schemas import (
            StartPromptEventSchema,
            FinishPromptEventSchema,
            StartTaskEventSchema,
            FinishTaskEventSchema,
            StartActionSubtaskEventSchema,
            FinishActionSubtaskEventSchema,
            StartStructureRunEventSchema,
            FinishStructureRunEventSchema,
            CompletionChunkEventSchema,
        )

        class_registry.register("StartPromptEvent", StartPromptEventSchema)
        class_registry.register("FinishPromptEvent", FinishPromptEventSchema)
        class_registry.register("StartTaskEvent", StartTaskEventSchema)
        class_registry.register("FinishTaskEvent", FinishTaskEventSchema)
        class_registry.register("StartActionSubtaskEvent", StartActionSubtaskEventSchema)
        class_registry.register("FinishActionSubtaskEvent", FinishActionSubtaskEventSchema)
        class_registry.register("StartStructureRunEvent", StartStructureRunEventSchema)
        class_registry.register("FinishStructureRunEvent", FinishStructureRunEventSchema)
        class_registry.register("CompletionChunkEvent", CompletionChunkEventSchema)

        try:
            return class_registry.get_class(event_dict["type"])().load(event_dict)
        except RegistryError:
            raise ValueError("Unsupported event type")

    @classmethod
    def from_json(cls, artifact_str: str) -> BaseEvent:
        return cls.from_dict(json.loads(artifact_str))

    def __str__(self) -> str:
        return json.dumps(self.to_dict())

    def to_json(self) -> str:
        return json.dumps(self.to_dict())

    @abstractmethod
    def to_dict(self) -> dict:
        ...

timestamp: float = field(default=Factory(lambda : time.time()), kw_only=True) class-attribute instance-attribute

type: str = field(default=Factory(lambda : self.__class__.__name__, takes_self=True), kw_only=True) class-attribute instance-attribute

__str__()

Source code in griptape/griptape/events/base_event.py 
48
49
def __str__(self) -> str:
    return json.dumps(self.to_dict())

from_dict(event_dict) classmethod

Source code in griptape/griptape/events/base_event.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
@classmethod
def from_dict(cls, event_dict: dict) -> BaseEvent:
    from griptape.schemas import (
        StartPromptEventSchema,
        FinishPromptEventSchema,
        StartTaskEventSchema,
        FinishTaskEventSchema,
        StartActionSubtaskEventSchema,
        FinishActionSubtaskEventSchema,
        StartStructureRunEventSchema,
        FinishStructureRunEventSchema,
        CompletionChunkEventSchema,
    )

    class_registry.register("StartPromptEvent", StartPromptEventSchema)
    class_registry.register("FinishPromptEvent", FinishPromptEventSchema)
    class_registry.register("StartTaskEvent", StartTaskEventSchema)
    class_registry.register("FinishTaskEvent", FinishTaskEventSchema)
    class_registry.register("StartActionSubtaskEvent", StartActionSubtaskEventSchema)
    class_registry.register("FinishActionSubtaskEvent", FinishActionSubtaskEventSchema)
    class_registry.register("StartStructureRunEvent", StartStructureRunEventSchema)
    class_registry.register("FinishStructureRunEvent", FinishStructureRunEventSchema)
    class_registry.register("CompletionChunkEvent", CompletionChunkEventSchema)

    try:
        return class_registry.get_class(event_dict["type"])().load(event_dict)
    except RegistryError:
        raise ValueError("Unsupported event type")

from_json(artifact_str) classmethod

Source code in griptape/griptape/events/base_event.py 
44
45
46
@classmethod
def from_json(cls, artifact_str: str) -> BaseEvent:
    return cls.from_dict(json.loads(artifact_str))

to_dict() abstractmethod

Source code in griptape/griptape/events/base_event.py 
54
55
56
@abstractmethod
def to_dict(self) -> dict:
    ...

to_json()

Source code in griptape/griptape/events/base_event.py 
51
52
def to_json(self) -> str:
    return json.dumps(self.to_dict())

BaseTaskEvent

Bases: BaseEvent, ABC

Source code in griptape/griptape/events/base_task_event.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
@define
class BaseTaskEvent(BaseEvent, ABC):
    task_id: str = field(kw_only=True)
    task_parent_ids: list[str] = field(kw_only=True)
    task_child_ids: list[str] = field(kw_only=True)

    task_input: BaseArtifact = field(kw_only=True)
    task_output: Optional[BaseArtifact] = field(kw_only=True)

    @classmethod
    def from_task(cls, task: BaseTask) -> BaseTaskEvent:
        return cls(
            task_id=task.id,
            task_parent_ids=task.parent_ids,
            task_child_ids=task.child_ids,
            task_input=task.input,
            task_output=task.output,
        )

task_child_ids: list[str] = field(kw_only=True) class-attribute instance-attribute

task_id: str = field(kw_only=True) class-attribute instance-attribute

task_input: BaseArtifact = field(kw_only=True) class-attribute instance-attribute

task_output: Optional[BaseArtifact] = field(kw_only=True) class-attribute instance-attribute

task_parent_ids: list[str] = field(kw_only=True) class-attribute instance-attribute

from_task(task) classmethod

Source code in griptape/griptape/events/base_task_event.py 
21
22
23
24
25
26
27
28
29
@classmethod
def from_task(cls, task: BaseTask) -> BaseTaskEvent:
    return cls(
        task_id=task.id,
        task_parent_ids=task.parent_ids,
        task_child_ids=task.child_ids,
        task_input=task.input,
        task_output=task.output,
    )

CompletionChunkEvent

Bases: BaseEvent

Source code in griptape/griptape/events/completion_chunk_event.py 
 6
 7
 8
 9
10
11
12
13
@define
class CompletionChunkEvent(BaseEvent):
    token: str = field(kw_only=True)

    def to_dict(self):
        from griptape.schemas import CompletionChunkEventSchema

        return dict(CompletionChunkEventSchema().dump(self))

token: str = field(kw_only=True) class-attribute instance-attribute

to_dict()

Source code in griptape/griptape/events/completion_chunk_event.py 
10
11
12
13
def to_dict(self):
    from griptape.schemas import CompletionChunkEventSchema

    return dict(CompletionChunkEventSchema().dump(self))

EventListener

Source code in griptape/griptape/events/event_listener.py 
6
7
8
9
@define
class EventListener:
    handler: Callable[[BaseEvent], Any] = field()
    event_types: Optional[list[Type[BaseEvent]]] = field(default=None, kw_only=True)

event_types: Optional[list[Type[BaseEvent]]] = field(default=None, kw_only=True) class-attribute instance-attribute

handler: Callable[[BaseEvent], Any] = field() class-attribute instance-attribute

FinishActionSubtaskEvent

Bases: BaseActionSubtaskEvent

Source code in griptape/griptape/events/finish_action_subtask_event.py 
 6
 7
 8
 9
10
11
@define
class FinishActionSubtaskEvent(BaseActionSubtaskEvent):
    def to_dict(self) -> dict:
        from griptape.schemas import FinishActionSubtaskEventSchema

        return dict(FinishActionSubtaskEventSchema().dump(self))

to_dict()

Source code in griptape/griptape/events/finish_action_subtask_event.py 
 8
 9
10
11
def to_dict(self) -> dict:
    from griptape.schemas import FinishActionSubtaskEventSchema

    return dict(FinishActionSubtaskEventSchema().dump(self))

FinishPromptEvent

Bases: BaseEvent

Source code in griptape/griptape/events/finish_prompt_event.py 
 6
 7
 8
 9
10
11
12
13
@define
class FinishPromptEvent(BaseEvent):
    token_count: int = field(kw_only=True)

    def to_dict(self) -> dict:
        from griptape.schemas import FinishPromptEventSchema

        return dict(FinishPromptEventSchema().dump(self))

token_count: int = field(kw_only=True) class-attribute instance-attribute

to_dict()

Source code in griptape/griptape/events/finish_prompt_event.py 
10
11
12
13
def to_dict(self) -> dict:
    from griptape.schemas import FinishPromptEventSchema

    return dict(FinishPromptEventSchema().dump(self))

FinishStructureRunEvent

Bases: BaseEvent

Source code in griptape/griptape/events/finish_structure_run_event.py 
 5
 6
 7
 8
 9
10
@define
class FinishStructureRunEvent(BaseEvent):
    def to_dict(self) -> dict:
        from griptape.schemas import FinishStructureRunEventSchema

        return dict(FinishStructureRunEventSchema().dump(self))

to_dict()

Source code in griptape/griptape/events/finish_structure_run_event.py 
 7
 8
 9
10
def to_dict(self) -> dict:
    from griptape.schemas import FinishStructureRunEventSchema

    return dict(FinishStructureRunEventSchema().dump(self))

FinishTaskEvent

Bases: BaseTaskEvent

Source code in griptape/griptape/events/finish_task_event.py 
 6
 7
 8
 9
10
11
@define
class FinishTaskEvent(BaseTaskEvent):
    def to_dict(self) -> dict:
        from griptape.schemas import FinishTaskEventSchema

        return dict(FinishTaskEventSchema().dump(self))

to_dict()

Source code in griptape/griptape/events/finish_task_event.py 
 8
 9
10
11
def to_dict(self) -> dict:
    from griptape.schemas import FinishTaskEventSchema

    return dict(FinishTaskEventSchema().dump(self))

StartActionSubtaskEvent

Bases: BaseActionSubtaskEvent

Source code in griptape/griptape/events/start_action_subtask_event.py 
 6
 7
 8
 9
10
11
@define
class StartActionSubtaskEvent(BaseActionSubtaskEvent):
    def to_dict(self) -> dict:
        from griptape.schemas import StartActionSubtaskEventSchema

        return dict(StartActionSubtaskEventSchema().dump(self))

to_dict()

Source code in griptape/griptape/events/start_action_subtask_event.py 
 8
 9
10
11
def to_dict(self) -> dict:
    from griptape.schemas import StartActionSubtaskEventSchema

    return dict(StartActionSubtaskEventSchema().dump(self))

StartPromptEvent

Bases: BaseEvent

Source code in griptape/griptape/events/start_prompt_event.py 
 5
 6
 7
 8
 9
10
11
12
@define
class StartPromptEvent(BaseEvent):
    token_count: int = field(kw_only=True)

    def to_dict(self) -> dict:
        from griptape.schemas import StartPromptEventSchema

        return dict(StartPromptEventSchema().dump(self))

token_count: int = field(kw_only=True) class-attribute instance-attribute

to_dict()

Source code in griptape/griptape/events/start_prompt_event.py 
 9
10
11
12
def to_dict(self) -> dict:
    from griptape.schemas import StartPromptEventSchema

    return dict(StartPromptEventSchema().dump(self))

StartStructureRunEvent

Bases: BaseEvent

Source code in griptape/griptape/events/start_structure_run_event.py 
 5
 6
 7
 8
 9
10
@define
class StartStructureRunEvent(BaseEvent):
    def to_dict(self) -> dict:
        from griptape.schemas import StartStructureRunEventSchema

        return dict(StartStructureRunEventSchema().dump(self))

to_dict()

Source code in griptape/griptape/events/start_structure_run_event.py 
 7
 8
 9
10
def to_dict(self) -> dict:
    from griptape.schemas import StartStructureRunEventSchema

    return dict(StartStructureRunEventSchema().dump(self))

StartTaskEvent

Bases: BaseTaskEvent

Source code in griptape/griptape/events/start_task_event.py 
 6
 7
 8
 9
10
11
@define
class StartTaskEvent(BaseTaskEvent):
    def to_dict(self) -> dict:
        from griptape.schemas import StartTaskEventSchema

        return dict(StartTaskEventSchema().dump(self))

to_dict()

Source code in griptape/griptape/events/start_task_event.py 
 8
 9
10
11
def to_dict(self) -> dict:
    from griptape.schemas import StartTaskEventSchema

    return dict(StartTaskEventSchema().dump(self))