Events

Events#

Overview#

You can configure the global EventBus with :ref:``EventListenerDriver.on_event` Return Value Behavior <EventListenerDriver.on_event Return Value Behavior>` s to listen for various framework events. See Event Listener Drivers for examples on forwarding events to external services.

Specific Event Types#

You can listen to specific event types:

from griptape.events import (
    BaseEvent,
    EventBus,
    EventListener,
    FinishActionsSubtaskEvent,
    FinishPromptEvent,
    FinishTaskEvent,
    StartActionsSubtaskEvent,
    StartPromptEvent,
    StartTaskEvent,
)
from griptape.structures import Agent

def on_event(event: BaseEvent) -> None:
    print(event.__class__)

EventBus.add_event_listeners(
    [
        EventListener(
            on_event,
            event_types=[
                StartTaskEvent,
                FinishTaskEvent,
                StartActionsSubtaskEvent,
                FinishActionsSubtaskEvent,
                StartPromptEvent,
                FinishPromptEvent,
            ],
        )
    ]
)

agent = Agent()

agent.run("tell me about griptape")

All Event Types#

Or listen to all events:

from griptape.events import BaseEvent, EventBus, EventListener
from griptape.structures import Agent

def handler1(event: BaseEvent) -> None:
    print("Handler 1", event.__class__)

def handler2(event: BaseEvent) -> None:
    print("Handler 2", event.__class__)

EventBus.add_event_listeners(
    [
        EventListener(handler1),
        EventListener(handler2),
    ]
)

agent = Agent()

agent.run("tell me about griptape")

Stream Iterator#

You can use Structure.run_stream() for streaming Events from the Structure in the form of an iterator.

Tip

Set stream=True on your Prompt Drivers in order to receive completion chunk events.

from griptape.events import BaseEvent
from griptape.structures import Agent

agent = Agent()

for event in agent.run_stream("Hi!", event_types=[BaseEvent]):  # All Events
    print(type(event))

Context Managers#

You can also use :ref:EventListenerDriver.on_event`  Return Value Behavior <`EventListenerDriver.on_event`  Return Value Behavior>`  s as a Python Context Manager. The ``EventListener will automatically be added and removed from the EventBus when entering and exiting the context.

from griptape.events import EventBus, EventListener, FinishStructureRunEvent, StartPromptEvent
from griptape.structures import Agent

EventBus.add_event_listeners(
    [EventListener(lambda e: print(f"Out of context: {e.type}"), event_types=[StartPromptEvent])]
)

agent = Agent(input="Hello!")

with EventListener(lambda e: print(f"In context: {e.type}"), event_types=[FinishStructureRunEvent]):
    agent.run()

agent.run()

Streaming#

You can use the BaseChunkEvent to stream the completion results from Prompt Drivers.

from griptape.drivers.prompt.openai import OpenAiChatPromptDriver
from griptape.events import BaseChunkEvent, EventBus, EventListener
from griptape.structures import Pipeline
from griptape.tasks import PromptTask
from griptape.tools import PromptSummaryTool, WebScraperTool

EventBus.add_event_listeners(
    [
        EventListener(
            lambda e: print(str(e), end="", flush=True),
            event_types=[BaseChunkEvent],
        ),
    ]
)

pipeline = Pipeline()
pipeline.add_tasks(
    PromptTask(
        "Based on https://griptape.ai, tell me what griptape is.",
        prompt_driver=OpenAiChatPromptDriver(model="gpt-4.1", stream=True),
        tools=[WebScraperTool(off_prompt=True), PromptSummaryTool(off_prompt=False)],
    )
)

pipeline.run()

You can also use the TextChunkEvent and ActionChunkEvent to further differentiate the different types of chunks for more customized output.

from griptape.drivers.prompt.openai import OpenAiChatPromptDriver
from griptape.events import ActionChunkEvent, EventBus, EventListener, TextChunkEvent
from griptape.structures import Pipeline
from griptape.tasks import PromptTask
from griptape.tools import PromptSummaryTool, WebScraperTool

EventBus.add_event_listeners(
    [
        EventListener(
            lambda e: print(str(e), end="", flush=True),
            event_types=[TextChunkEvent],
        ),
        EventListener(
            lambda e: print(str(e), end="", flush=True),
            event_types=[ActionChunkEvent],
        ),
    ]
)

pipeline = Pipeline()
pipeline.add_tasks(
    PromptTask(
        "Based on https://griptape.ai, tell me what griptape is.",
        prompt_driver=OpenAiChatPromptDriver(model="gpt-4.1", stream=True),
        tools=[WebScraperTool(off_prompt=True), PromptSummaryTool(off_prompt=False)],
    )
)

pipeline.run()

If you want Gen AI Builder to handle the chunk events for you, use the Stream Iterator utility to automatically wrap BaseChunkEvent s in a Python iterator.

The Stream utility does not automatically enable streaming on the Drivers that produce BaseChunkEvent s. Make sure to enable streaming on the Drivers or else Stream will yield no iterations.

import logging

from griptape.configs import Defaults
from griptape.structures import Agent
from griptape.tools import PromptSummaryTool, WebScraperTool
from griptape.utils import Stream

#  Hide Griptapes usual output

logging.getLogger(Defaults.logging_config.logger_name).setLevel(logging.ERROR)

agent = Agent(
    input="Based on https://griptape.ai, tell me what griptape is.",
    tools=[
        PromptSummaryTool(off_prompt=True),
        WebScraperTool(off_prompt=False),
    ],
    stream=True,
)

for artifact in Stream(agent).run():
    print(artifact.value, end="", flush=True)

Sometimes, streaming can be too verbose. You can use Stream.event_types to only listen to specific event types. A good example is to remove the ActionChunkEvent from the stream if you don’t need to see events related to Tool usage.

import logging

from griptape.configs import Defaults
from griptape.events import FinishPromptEvent, FinishStructureRunEvent, TextChunkEvent
from griptape.structures import Agent
from griptape.tools import PromptSummaryTool, WebScraperTool
from griptape.utils import Stream

#  Hide Griptapes usual output

logging.getLogger(Defaults.logging_config.logger_name).setLevel(logging.ERROR)

agent = Agent(
    input="Based on https://griptape.ai, tell me what griptape is.",
    tools=[
        PromptSummaryTool(off_prompt=True),
        WebScraperTool(off_prompt=False),
    ],
    stream=True,
)

#  Listen for the following event types

event_types = [TextChunkEvent, FinishPromptEvent, FinishStructureRunEvent]

for artifact in Stream(agent, event_types=event_types).run():
    print(artifact.value, end="", flush=True)

Counting Tokens#

To count tokens, you can use Event Listeners and the TokenCounter util:

- -8<-- "docs/griptape-framework/misc/src/events_5.py"

Inspecting Payloads#

You can use the StartPromptEvent to inspect the Prompt Stack and final prompt string before it is sent to the LLM.

- -8<-- "docs/griptape-framework/misc/src/events_6.py"

...
Prompt Stack Messages:
system:
user: Write me a poem.
Final Prompt String:

User: Write me a poem.

Assistant:
...

EventListenerDriver.on_event Return Value Behavior#

The value that gets returned from the :ref:EventListener.on_event`  is None <`EventListener.on_event`  is None>`  will determine what gets sent to the ``event_listener_driver .

EventListener.on_event is None#

By default, the EventListener.on_event function is None . Any events that the EventListener is listening for will get sent to the event_listener_driver as-is.

Return BaseEvent or dict#

You can return a BaseEvent or dict object from EventListener.on_event , and it will get sent to the event_listener_driver .

Return None#

You can return None in the on_event function to prevent the event from getting sent to the event_listener_driver .

- -8<-- "docs/griptape-framework/misc/src/events_no_publish.py"
Contents