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 contentevent_type
str, optional - Filter by event type (e.g., "FileReadAction")source
str, optional - Filter by event sourcestart_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 0limit
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