openhands.events.stream

EventStream Objects

@dataclass
class EventStream()

get_events

def get_events(start_id: int = 0,
               end_id: int | None = None,
               reverse: bool = False,
               filter_out_type: tuple[type[Event], ...] | None = None,
               filter_hidden=False) -> Iterable[Event]

Retrieve events from the event stream, optionally filtering out events of a given type and events marked as hidden.

Arguments:

• start_id - The ID of the first event to retrieve. Defaults to 0.
• end_id - The ID of the last event to retrieve. Defaults to the last event in the stream.
• reverse - Whether to retrieve events in reverse order. Defaults to False.
• filter_out_type - A tuple of event types to filter out. Typically used to filter out backend events from the agent.
• filter_hidden - If True, filters out events with the 'hidden' attribute set to True.

Yields:

Events from the stream that match the criteria.

get_matching_events

def get_matching_events(query: str | None = None,
                        event_type: str | None = None,
                        source: str | None = None,
                        start_date: str | None = None,
                        end_date: str | None = None,
                        start_id: int = 0,
                        limit: int = 100) -> list

Get matching events from the event stream based on filters.

Arguments:

• query str, optional - Text to search for in event content
• event_type str, optional - Filter by event type (e.g., "FileReadAction")
• source str, optional - Filter by event source
• start_date str, optional - Filter events after this date (ISO format)
• end_date str, optional - Filter events before this date (ISO format)
• start_id int - Starting ID in the event stream. Defaults to 0
• limit int - Maximum number of events to return. Must be between 1 and 100. Defaults to 100

Returns:

• list - List of matching events (as dicts)

Raises:

• ValueError - If limit is less than 1 or greater than 100