events

events#

  • EventBus = _EventBus() module-attribute

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

Bases: BaseChunkEvent “)

Source Code in griptape/events/action_chunk_event.py

@define
class ActionChunkEvent(BaseChunkEvent):
    partial_input: Optional[str] = field(default=None, kw_only=True, metadata={"serializable": True})
    tag: Optional[str] = field(default=None, kw_only=True, metadata={"serializable": True})
    name: Optional[str] = field(default=None, kw_only=True, metadata={"serializable": True})
    path: Optional[str] = field(default=None, kw_only=True, metadata={"serializable": True})

    def __str__(self) -> str:
        parts = []

        if self.name:
            parts.append(self.name)
            if self.path:
                parts.append(f".{self.path}")
                if self.tag:
                    parts.append(f" ({self.tag})")

        if self.partial_input:
            if parts:
                parts.append(f"\n{self.partial_input}")
            else:
                parts.append(self.partial_input)

        return "".join(parts)

  • name = field(default=None, kw_only=True, metadata={'serializable': True}) class-attribute instance-attribute

  • partial_input = field(default=None, kw_only=True, metadata={'serializable': True}) class-attribute instance-attribute

  • path = field(default=None, kw_only=True, metadata={'serializable': True}) class-attribute instance-attribute

  • tag = field(default=None, kw_only=True, metadata={'serializable': True}) class-attribute instance-attribute

str()#

Source Code in griptape/events/action_chunk_event.py

def __str__(self) -> str:
    parts = []

    if self.name:
        parts.append(self.name)
        if self.path:
            parts.append(f".{self.path}")
            if self.tag:
                parts.append(f" ({self.tag})")

    if self.partial_input:
        if parts:
            parts.append(f"\n{self.partial_input}")
        else:
            parts.append(self.partial_input)

    return "".join(parts)

AudioChunkEvent#

Bases: BaseChunkEvent “)

Attributes#

Name

Type

Description

<a href="./#griptape.events.AudioChunkEvent.data" title="data = field(kw_only=True, metadata={'serializable': True}) class-attribute instance-attribute (griptape.events.AudioChunkEvent.data)"><code class="language-python">data</code></a>

str

Base64 encoded audio data.<br/>

Source Code in griptape/events/audio_chunk_event.py

@define
class AudioChunkEvent(BaseChunkEvent):
    """Stores a chunk of audio data.

    Attributes:
        data: Base64 encoded audio data.
    """

    data: str = field(kw_only=True, metadata={"serializable": True})

    def __str__(self) -> str:
        return self.data

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

str()#

Source Code in griptape/events/audio_chunk_event.py

def __str__(self) -> str:
    return self.data

BaseActionsSubtaskEvent#

Bases: BaseTaskEvent “) , ABC

Source Code in griptape/events/base_actions_subtask_event.py

@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.origin_task.id,
            subtask_thought=task.thought,
            subtask_actions=task.actions_to_dicts(),
        )

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

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

  • subtask_thought = 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

@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.origin_task.id,
        subtask_thought=task.thought,
        subtask_actions=task.actions_to_dicts(),
    )

BaseAudioTranscriptionEvent#

Bases: Return `BaseEvent or dict <Return BaseEvent or dict>` “)

Source Code in griptape/events/base_audio_transcription_event.py

class BaseAudioTranscriptionEvent(BaseEvent): ...

BaseChunkEvent#

Bases: Return `BaseEvent or dict <Return BaseEvent or dict>` “)

Source Code in griptape/events/base_chunk_event.py

@define
class BaseChunkEvent(BaseEvent):
    index: int = field(default=0, metadata={"serializable": True})

    @abstractmethod
    def __str__(self) -> str: ...

  • index = field(default=0, metadata={'serializable': True}) class-attribute instance-attribute

str()abstractmethod#

Source Code in griptape/events/base_chunk_event.py

@abstractmethod
def __str__(self) -> str: ...

BaseEvent#

Bases: SerializableMixin “) , ABC

Source Code in griptape/events/base_event.py

@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 = field(default=Factory(lambda: uuid.uuid4().hex), kw_only=True, metadata={'serializable': True}) class-attribute instance-attribute

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

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

BasePromptEvent#

Bases: Return `BaseEvent or dict <Return BaseEvent or dict>` “) , ABC

Source Code in griptape/events/base_prompt_event.py

@define
class BasePromptEvent(BaseEvent, ABC):
    model: str = field(kw_only=True, metadata={"serializable": True})

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

BaseTaskEvent#

Bases: Return `BaseEvent or dict <Return BaseEvent or dict>` “) , ABC

Source Code in griptape/events/base_task_event.py

@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 = field(kw_only=True, metadata={'serializable': True}) class-attribute instance-attribute

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

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

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

  • task_parent_ids = 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

@define
class BaseTextToSpeechEvent(BaseMediaGenerationEvent, ABC): ...

EventListener#

Bases: Generic[T

Attributes#

Name

Type

Description

<a href="./#griptape.events.EventListener.on_event" title="on_event = field(default=None) class-attribute instance-attribute (griptape.events.EventListener.on_event)"><code class="language-python">on_event</code></a>

<a href="event_listener.mdx#griptape.events.event_listener.T" title="T = TypeVar('T', bound=BaseEvent) module-attribute (griptape.events.event_listener.T)"><code class="language-python">Optional[Callable[[T], Optional[BaseEvent I dict]]]</code></a>

The on_event function that will be called when an event is published. The on_event function should accept an event and return either the event or a dictionary. If the on_event returns None, the event will not be published.<br/>

<a href="./#griptape.events.EventListener.event_types" title="event_types = field(default=None, kw_only=True) class-attribute instance-attribute (griptape.events.EventListener.event_types)"><code class="language-python">event_types</code></a>

<a href="../../data/artifacts.mdx#list" title="List (list)"><code class="language-python">Optional[list[type[T]]]</code></a>

A list of event types that the event listener should listen for. If not provided, the event listener will listen for all event types.<br/>

<a href="./#griptape.events.EventListener.event_listener_driver" title="event_listener_driver = field(default=None, kw_only=True) class-attribute instance-attribute (griptape.events.EventListener.event_listener_driver)"><code class="language-python">event_listener_driver</code></a>

<a href="../drivers/event_listener#griptape.drivers.event_listener.BaseEventListenerDriver" title="BaseEventListenerDriver (griptape.drivers.event_listener.BaseEventListenerDriver)"><code class="language-python">Optional[BaseEventListenerDriver]</code></a>

The driver that will be used to publish events.<br/>

Source Code in griptape/events/event_listener.py

@define
class EventListener(Generic[T]):
    """An event listener that listens for events and handles them.

    Attributes:
        on_event: The on_event function that will be called when an event is published.
            The on_event function should accept an event and return either the event or a dictionary.
            If the on_event returns None, the event will not be published.
        event_types: A list of event types that the event listener should listen for.
            If not provided, the event listener will listen for all event types.
        event_listener_driver: The driver that will be used to publish events.
    """

    on_event: Optional[Callable[[T], Optional[BaseEvent | dict]]] = field(default=None)
    event_types: Optional[list[type[T]]] = field(default=None, kw_only=True)
    event_listener_driver: Optional[BaseEventListenerDriver] = field(default=None, kw_only=True)

    def __enter__(self) -> Self:
        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)

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

        if event_types is None or any(isinstance(event, event_type) for event_type in event_types):
            handled_event = event
            if self.on_event is not None:
                handled_event = self.on_event(event)

            if self.event_listener_driver is not None and handled_event is not None:
                self.event_listener_driver.publish_event(handled_event)

        if self.event_listener_driver is not None and flush:
            self.event_listener_driver.flush_events()

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

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

  • on_event = field(default=None) class-attribute instance-attribute

enter()#

Source Code in griptape/events/event_listener.py

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

    EventBus.add_event_listener(self)

    return self

exit(type, value, traceback)#

Source Code in griptape/events/event_listener.py

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

    EventBus.remove_event_listener(self)

publish_event(event, *, flush=False)#

Source Code in griptape/events/event_listener.py

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

    if event_types is None or any(isinstance(event, event_type) for event_type in event_types):
        handled_event = event
        if self.on_event is not None:
            handled_event = self.on_event(event)

        if self.event_listener_driver is not None and handled_event is not None:
            self.event_listener_driver.publish_event(handled_event)

    if self.event_listener_driver is not None and flush:
        self.event_listener_driver.flush_events()

FinishActionsSubtaskEvent#

Bases: BaseActionsSubtaskEvent “)

Source Code in griptape/events/finish_actions_subtask_event.py

@define
class FinishActionsSubtaskEvent(BaseActionsSubtaskEvent): ...

FinishAudioTranscriptionEvent#

Bases: BaseAudioTranscriptionEvent “)

Source Code in griptape/events/finish_audio_transcription_event.py

class FinishAudioTranscriptionEvent(BaseAudioTranscriptionEvent): ...

FinishImageGenerationEvent#

Bases: BaseImageGenerationEvent “)

Source Code in griptape/events/finish_image_generation_event.py

@define
class FinishImageGenerationEvent(BaseImageGenerationEvent): ...

FinishImageQueryEvent#

Bases: BaseImageQueryEvent “)

Source Code in griptape/events/finish_image_query_event.py

@define
class FinishImageQueryEvent(BaseImageQueryEvent):
    result: str = field(kw_only=True, metadata={"serializable": True})

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

FinishPromptEvent#

Bases: BasePromptEvent “)

Source Code in griptape/events/finish_prompt_event.py

@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 = field(kw_only=True, metadata={'serializable': True}) class-attribute instance-attribute

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

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

FinishStructureRunEvent#

Bases: Return `BaseEvent or dict <Return BaseEvent or dict>` “)

Source Code in griptape/events/finish_structure_run_event.py

@define
class FinishStructureRunEvent(BaseEvent):
    structure_id: Optional[str] = field(kw_only=True, default=None, metadata={"serializable": True})
    output_task_input: Optional[BaseArtifact] = field(kw_only=True, default=None, metadata={"serializable": True})
    output_task_output: Optional[BaseArtifact] = field(kw_only=True, default=None, metadata={"serializable": True})

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

  • output_task_output = field(kw_only=True, default=None, metadata={'serializable': True}) class-attribute instance-attribute

  • structure_id = 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

@define
class FinishTaskEvent(BaseTaskEvent): ...

FinishTextToSpeechEvent#

Bases: BaseTextToSpeechEvent “)

Source Code in griptape/events/finish_text_to_speech_event.py

@define
class FinishTextToSpeechEvent(BaseTextToSpeechEvent): ...

StartActionsSubtaskEvent#

Bases: BaseActionsSubtaskEvent “)

Source Code in griptape/events/start_actions_subtask_event.py

@define
class StartActionsSubtaskEvent(BaseActionsSubtaskEvent): ...

StartAudioTranscriptionEvent#

Bases: BaseAudioTranscriptionEvent “)

Source Code in griptape/events/start_audio_transcription_event.py

class StartAudioTranscriptionEvent(BaseAudioTranscriptionEvent): ...

StartImageGenerationEvent#

Bases: BaseImageGenerationEvent “)

Source Code in griptape/events/start_image_generation_event.py

@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 = field(default=None, kw_only=True, metadata={'serializable': True}) class-attribute instance-attribute

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

StartImageQueryEvent#

Bases: BaseImageQueryEvent “)

Source Code in griptape/events/start_image_query_event.py

@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 = field(kw_only=True, metadata={'serializable': True}) class-attribute instance-attribute

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

StartPromptEvent#

Bases: BasePromptEvent “)

Source Code in griptape/events/start_prompt_event.py

@define
class StartPromptEvent(BasePromptEvent):
    prompt_stack: PromptStack = field(kw_only=True, metadata={"serializable": True})

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

StartStructureRunEvent#

Bases: Return `BaseEvent or dict <Return BaseEvent or dict>` “)

Source Code in griptape/events/start_structure_run_event.py

@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 = field(kw_only=True, metadata={'serializable': True}) class-attribute instance-attribute

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

  • structure_id = 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

@define
class StartTaskEvent(BaseTaskEvent): ...

StartTextToSpeechEvent#

Bases: BaseTextToSpeechEvent “)

Source Code in griptape/events/start_text_to_speech_event.py

@define
class StartTextToSpeechEvent(BaseTextToSpeechEvent):
    prompts: list[str] = field(kw_only=True, metadata={"serializable": True})

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

TextChunkEvent#

Bases: BaseChunkEvent “)

Source Code in griptape/events/text_chunk_event.py

@define
class TextChunkEvent(BaseChunkEvent):
    token: str = field(kw_only=True, metadata={"serializable": True})

    def __str__(self) -> str:
        return self.token

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

str()#

Source Code in griptape/events/text_chunk_event.py

def __str__(self) -> str:
    return self.token
Contents