Skip to content

events

EventBus = _EventBus() module-attribute

__all__ = ['BaseEvent', 'BaseTaskEvent', 'StartTaskEvent', 'FinishTaskEvent', 'BaseActionsSubtaskEvent', 'StartActionsSubtaskEvent', 'FinishActionsSubtaskEvent', 'BasePromptEvent', 'StartPromptEvent', 'FinishPromptEvent', 'StartStructureRunEvent', 'FinishStructureRunEvent', 'CompletionChunkEvent', 'EventListener', 'StartImageGenerationEvent', 'FinishImageGenerationEvent', 'StartImageQueryEvent', 'FinishImageQueryEvent', 'BaseTextToSpeechEvent', 'StartTextToSpeechEvent', 'FinishTextToSpeechEvent', 'BaseAudioTranscriptionEvent', 'StartAudioTranscriptionEvent', 'FinishAudioTranscriptionEvent', 'EventBus'] module-attribute

BaseActionsSubtaskEvent

Bases: BaseTaskEvent, ABC

Source code in griptape/events/base_actions_subtask_event.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
@define
class BaseActionsSubtaskEvent(BaseTaskEvent, ABC):
    subtask_parent_task_id: Optional[str] = field(kw_only=True, metadata={"serializable": True})
    subtask_thought: Optional[str] = field(kw_only=True, metadata={"serializable": True})
    subtask_actions: Optional[list[dict]] = field(kw_only=True, metadata={"serializable": True})

    @classmethod
    def from_task(cls, task: BaseTask) -> BaseActionsSubtaskEvent:
        from griptape.tasks import ActionsSubtask

        if not isinstance(task, ActionsSubtask):
            raise ValueError("Event must be of instance ActionSubtask.")
        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_actions=task.actions_to_dicts(),
        )

subtask_actions: Optional[list[dict]] = field(kw_only=True, metadata={'serializable': True}) class-attribute instance-attribute

subtask_parent_task_id: Optional[str] = field(kw_only=True, metadata={'serializable': True}) class-attribute instance-attribute

subtask_thought: Optional[str] = field(kw_only=True, metadata={'serializable': True}) class-attribute instance-attribute

from_task(task) classmethod

Source code in griptape/events/base_actions_subtask_event.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
@classmethod
def from_task(cls, task: BaseTask) -> BaseActionsSubtaskEvent:
    from griptape.tasks import ActionsSubtask

    if not isinstance(task, ActionsSubtask):
        raise ValueError("Event must be of instance ActionSubtask.")
    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_actions=task.actions_to_dicts(),
    )

BaseAudioTranscriptionEvent

Bases: BaseEvent

Source code in griptape/events/base_audio_transcription_event.py 
4
class BaseAudioTranscriptionEvent(BaseEvent): ...

BaseEvent

Bases: SerializableMixin, ABC

Source code in griptape/events/base_event.py 
13
14
15
16
17
@define
class BaseEvent(SerializableMixin, ABC):
    id: str = field(default=Factory(lambda: uuid.uuid4().hex), kw_only=True, metadata={"serializable": True})
    timestamp: float = field(default=Factory(lambda: time.time()), kw_only=True, metadata={"serializable": True})
    meta: dict[str, Any] = field(factory=dict, kw_only=True, metadata={"serializable": True})

id: str = field(default=Factory(lambda: uuid.uuid4().hex), kw_only=True, metadata={'serializable': True}) class-attribute instance-attribute

meta: dict[str, Any] = field(factory=dict, kw_only=True, metadata={'serializable': True}) class-attribute instance-attribute

timestamp: float = field(default=Factory(lambda: time.time()), kw_only=True, metadata={'serializable': True}) class-attribute instance-attribute

BasePromptEvent

Bases: BaseEvent, ABC

Source code in griptape/events/base_prompt_event.py 
10
11
12
@define
class BasePromptEvent(BaseEvent, ABC):
    model: str = field(kw_only=True, metadata={"serializable": True})

model: str = field(kw_only=True, metadata={'serializable': True}) class-attribute instance-attribute

BaseTaskEvent

Bases: BaseEvent, ABC

Source code in griptape/events/base_task_event.py 
14
15
16
17
18
19
20
21
@define
class BaseTaskEvent(BaseEvent, ABC):
    task_id: str = field(kw_only=True, metadata={"serializable": True})
    task_parent_ids: list[str] = field(kw_only=True, metadata={"serializable": True})
    task_child_ids: list[str] = field(kw_only=True, metadata={"serializable": True})

    task_input: BaseArtifact = field(kw_only=True, metadata={"serializable": True})
    task_output: Optional[BaseArtifact] = field(kw_only=True, metadata={"serializable": True})

task_child_ids: list[str] = field(kw_only=True, metadata={'serializable': True}) class-attribute instance-attribute

task_id: str = field(kw_only=True, metadata={'serializable': True}) class-attribute instance-attribute

task_input: BaseArtifact = field(kw_only=True, metadata={'serializable': True}) class-attribute instance-attribute

task_output: Optional[BaseArtifact] = field(kw_only=True, metadata={'serializable': True}) class-attribute instance-attribute

task_parent_ids: list[str] = field(kw_only=True, metadata={'serializable': True}) class-attribute instance-attribute

BaseTextToSpeechEvent

Bases: BaseMediaGenerationEvent, ABC

Source code in griptape/events/base_text_to_speech_event.py 
10
11
@define
class BaseTextToSpeechEvent(BaseMediaGenerationEvent, ABC): ...

CompletionChunkEvent

Bases: BaseEvent

Source code in griptape/events/completion_chunk_event.py 
6
7
8
@define
class CompletionChunkEvent(BaseEvent):
    token: str = field(kw_only=True, metadata={"serializable": True})

token: str = field(kw_only=True, metadata={'serializable': True}) class-attribute instance-attribute

EventListener

Source code in griptape/events/event_listener.py 
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
@define
class EventListener:
    handler: Callable[[BaseEvent], Optional[dict]] = field(default=Factory(lambda: lambda event: event.to_dict()))
    event_types: Optional[list[type[BaseEvent]]] = field(default=None, kw_only=True)
    driver: Optional[BaseEventListenerDriver] = field(default=None, kw_only=True)

    _last_event_listeners: Optional[list[EventListener]] = field(default=None)

    def __enter__(self) -> EventListener:
        from griptape.events import EventBus

        EventBus.add_event_listener(self)

        return self

    def __exit__(self, type, value, traceback) -> None:  # noqa: ANN001, A002
        from griptape.events import EventBus

        EventBus.remove_event_listener(self)

        self._last_event_listeners = None

    def publish_event(self, event: BaseEvent, *, flush: bool = False) -> None:
        event_types = self.event_types

        if event_types is None or type(event) in event_types:
            event_payload = self.handler(event)
            if self.driver is not None:
                if event_payload is not None and isinstance(event_payload, dict):
                    self.driver.publish_event(event_payload, flush=flush)
                else:
                    self.driver.publish_event(event, flush=flush)

driver: Optional[BaseEventListenerDriver] = field(default=None, kw_only=True) class-attribute instance-attribute

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

handler: Callable[[BaseEvent], Optional[dict]] = field(default=Factory(lambda: lambda event: event.to_dict())) class-attribute instance-attribute

__enter__()

Source code in griptape/events/event_listener.py 
21
22
23
24
25
26
def __enter__(self) -> EventListener:
    from griptape.events import EventBus

    EventBus.add_event_listener(self)

    return self

__exit__(type, value, traceback)

Source code in griptape/events/event_listener.py 
28
29
30
31
32
33
def __exit__(self, type, value, traceback) -> None:  # noqa: ANN001, A002
    from griptape.events import EventBus

    EventBus.remove_event_listener(self)

    self._last_event_listeners = None

publish_event(event, *, flush=False)

Source code in griptape/events/event_listener.py 
35
36
37
38
39
40
41
42
43
44
def publish_event(self, event: BaseEvent, *, flush: bool = False) -> None:
    event_types = self.event_types

    if event_types is None or type(event) in event_types:
        event_payload = self.handler(event)
        if self.driver is not None:
            if event_payload is not None and isinstance(event_payload, dict):
                self.driver.publish_event(event_payload, flush=flush)
            else:
                self.driver.publish_event(event, flush=flush)

FinishActionsSubtaskEvent

Bases: BaseActionsSubtaskEvent

Source code in griptape/events/finish_actions_subtask_event.py 
8
9
@define
class FinishActionsSubtaskEvent(BaseActionsSubtaskEvent): ...

FinishAudioTranscriptionEvent

Bases: BaseAudioTranscriptionEvent

Source code in griptape/events/finish_audio_transcription_event.py 
4
class FinishAudioTranscriptionEvent(BaseAudioTranscriptionEvent): ...

FinishImageGenerationEvent

Bases: BaseImageGenerationEvent

Source code in griptape/events/finish_image_generation_event.py 
8
9
@define
class FinishImageGenerationEvent(BaseImageGenerationEvent): ...

FinishImageQueryEvent

Bases: BaseImageQueryEvent

Source code in griptape/events/finish_image_query_event.py 
6
7
8
@define
class FinishImageQueryEvent(BaseImageQueryEvent):
    result: str = field(kw_only=True, metadata={"serializable": True})

result: str = field(kw_only=True, metadata={'serializable': True}) class-attribute instance-attribute

FinishPromptEvent

Bases: BasePromptEvent

Source code in griptape/events/finish_prompt_event.py 
10
11
12
13
14
@define
class FinishPromptEvent(BasePromptEvent):
    result: str = field(kw_only=True, metadata={"serializable": True})
    input_token_count: Optional[float] = field(kw_only=True, metadata={"serializable": True})
    output_token_count: Optional[float] = field(kw_only=True, metadata={"serializable": True})

input_token_count: Optional[float] = field(kw_only=True, metadata={'serializable': True}) class-attribute instance-attribute

output_token_count: Optional[float] = field(kw_only=True, metadata={'serializable': True}) class-attribute instance-attribute

result: str = field(kw_only=True, metadata={'serializable': True}) class-attribute instance-attribute

FinishStructureRunEvent

Bases: BaseEvent

Source code in griptape/events/finish_structure_run_event.py 
13
14
15
16
17
@define
class FinishStructureRunEvent(BaseEvent):
    structure_id: Optional[str] = field(kw_only=True, default=None, metadata={"serializable": True})
    output_task_input: BaseArtifact = field(kw_only=True, metadata={"serializable": True})
    output_task_output: Optional[BaseArtifact] = field(kw_only=True, metadata={"serializable": True})

output_task_input: BaseArtifact = field(kw_only=True, metadata={'serializable': True}) class-attribute instance-attribute

output_task_output: Optional[BaseArtifact] = field(kw_only=True, metadata={'serializable': True}) class-attribute instance-attribute

structure_id: Optional[str] = field(kw_only=True, default=None, metadata={'serializable': True}) class-attribute instance-attribute

FinishTaskEvent

Bases: BaseTaskEvent

Source code in griptape/events/finish_task_event.py 
8
9
@define
class FinishTaskEvent(BaseTaskEvent): ...

FinishTextToSpeechEvent

Bases: BaseTextToSpeechEvent

Source code in griptape/events/finish_text_to_speech_event.py 
8
9
@define
class FinishTextToSpeechEvent(BaseTextToSpeechEvent): ...

StartActionsSubtaskEvent

Bases: BaseActionsSubtaskEvent

Source code in griptape/events/start_actions_subtask_event.py 
8
9
@define
class StartActionsSubtaskEvent(BaseActionsSubtaskEvent): ...

StartAudioTranscriptionEvent

Bases: BaseAudioTranscriptionEvent

Source code in griptape/events/start_audio_transcription_event.py 
4
class StartAudioTranscriptionEvent(BaseAudioTranscriptionEvent): ...

StartImageGenerationEvent

Bases: BaseImageGenerationEvent

Source code in griptape/events/start_image_generation_event.py 
10
11
12
13
@define
class StartImageGenerationEvent(BaseImageGenerationEvent):
    prompts: list[str] = field(kw_only=True, metadata={"serializable": True})
    negative_prompts: Optional[list[str]] = field(default=None, kw_only=True, metadata={"serializable": True})

negative_prompts: Optional[list[str]] = field(default=None, kw_only=True, metadata={'serializable': True}) class-attribute instance-attribute

prompts: list[str] = field(kw_only=True, metadata={'serializable': True}) class-attribute instance-attribute

StartImageQueryEvent

Bases: BaseImageQueryEvent

Source code in griptape/events/start_image_query_event.py 
 8
 9
10
11
@define
class StartImageQueryEvent(BaseImageQueryEvent):
    query: str = field(kw_only=True, metadata={"serializable": True})
    images_info: list[str] = field(kw_only=True, metadata={"serializable": True})

images_info: list[str] = field(kw_only=True, metadata={'serializable': True}) class-attribute instance-attribute

query: str = field(kw_only=True, metadata={'serializable': True}) class-attribute instance-attribute

StartPromptEvent

Bases: BasePromptEvent

Source code in griptape/events/start_prompt_event.py 
13
14
15
@define
class StartPromptEvent(BasePromptEvent):
    prompt_stack: PromptStack = field(kw_only=True, metadata={"serializable": True})

prompt_stack: PromptStack = field(kw_only=True, metadata={'serializable': True}) class-attribute instance-attribute

StartStructureRunEvent

Bases: BaseEvent

Source code in griptape/events/start_structure_run_event.py 
13
14
15
16
17
@define
class StartStructureRunEvent(BaseEvent):
    structure_id: Optional[str] = field(kw_only=True, default=None, metadata={"serializable": True})
    input_task_input: BaseArtifact = field(kw_only=True, metadata={"serializable": True})
    input_task_output: Optional[BaseArtifact] = field(kw_only=True, metadata={"serializable": True})

input_task_input: BaseArtifact = field(kw_only=True, metadata={'serializable': True}) class-attribute instance-attribute

input_task_output: Optional[BaseArtifact] = field(kw_only=True, metadata={'serializable': True}) class-attribute instance-attribute

structure_id: Optional[str] = field(kw_only=True, default=None, metadata={'serializable': True}) class-attribute instance-attribute

StartTaskEvent

Bases: BaseTaskEvent

Source code in griptape/events/start_task_event.py 
8
9
@define
class StartTaskEvent(BaseTaskEvent): ...

StartTextToSpeechEvent

Bases: BaseTextToSpeechEvent

Source code in griptape/events/start_text_to_speech_event.py 
 8
 9
10
@define
class StartTextToSpeechEvent(BaseTextToSpeechEvent):
    prompts: list[str] = field(kw_only=True, metadata={"serializable": True})

prompts: list[str] = field(kw_only=True, metadata={'serializable': True}) class-attribute instance-attribute