How to Filter Event Results with Include/Exclude Predicates in bubus

bubus provides dual predicate mechanisms—include and exclude callables—that let you filter both incoming events via EventBus.expect and handler results via BaseEvent.event_results_filtered.

The browser-use/bubus event bus library offers powerful filtering capabilities that allow you to selectively process events and their outputs using simple predicate functions. Understanding how to filter event results with include/exclude predicates in bubus enables you to build precise asynchronous workflows that react only to relevant data. This guide examines the actual implementation in bubus/service.py and bubus/models.py to demonstrate exactly how these filtering mechanisms operate.

Understanding the Two Filtering Mechanisms

bubus implements predicate-based filtering at two distinct stages of the event lifecycle. Each mechanism serves a different purpose and operates on different data types.

Filtering Incoming Events with EventBus.expect

Located in bubus/service.py (lines 590-662), the EventBus.expect method creates a temporary handler that inspects every emitted event until a match is found. The implementation checks each event against your predicates before resolving the returned future:

def notify_expect_handler(event: BaseEvent[Any]) -> None:
    if not future.done() and include(event) and not exclude(event):
        future.set_result(event)

The handler registers with self.on(event_type, notify_expect_handler) and automatically unregisters once the future resolves or the timeout expires. This mechanism filters events before they are processed by your awaiting code.

Filtering Handler Results with event_results_filtered

After event handlers execute, they produce EventResult objects. The BaseEvent.event_results_filtered method, defined in bubus/models.py (lines 474-500), filters these completed results using the supplied include callable:

included_results = {
    handler_key: event_result
    for handler_key, event_result in event_results.items()
    if include(event_result)
}

By default, this method uses the private helper _event_result_is_truthy (lines 61-72), which returns True only for completed, non-error results that are not None or exception-wrapped.

Working with Include and Exclude Predicates

Both APIs accept two callable parameters: include (must return True for matches) and exclude (must return False for matches). When both are provided, an event or result must satisfy include AND NOT exclude to pass the filter.

For backward compatibility, both methods retain an optional predicate argument. According to the source code in bubus/service.py, if predicate is supplied, it is merged into include internally:

if predicate is not None:
    original_include = include
    include = lambda e, orig=original_include, pred=predicate: orig(e) and pred(e)

Practical Code Examples

The following examples demonstrate real-world usage patterns from the bubus test suite.

Filtering Events with Include Predicates

Wait for a specific ResponseEvent that matches a particular request_id:

response = await bus.expect(
    'ResponseEvent',
    include=lambda e: e.request_id == my_request_id,
    timeout=30,
)
print(response)  # → BaseEvent instance with matching request_id

Verified in tests/test_typed_event_results.py (lines 211-221), this pattern is ideal when you need to correlate responses with specific requests.

Excluding Unwanted Events

Prevent matches for events that contain error conditions:

response = await bus.expect(
    'ResponseEvent',
    exclude=lambda e: e.error_code is not None,
    timeout=30,
)

As shown in tests/test_eventbus.py (lines 648-650), the exclude predicate drops any event where the callable returns True.

Filtering Handler Results

Process only successful handler outputs that meet specific criteria:

filtered = await event.event_results_filtered(
    include=lambda er: isinstance(er.result, int) and er.result > 10,
)
print(filtered)  # → dict of handler_id → EventResult objects

This filters the EventResult objects after all handlers complete, keeping only integer results greater than 10.

Combining Include and Exclude Logic

Apply both predicates simultaneously to refine your results precisely:

filtered = await event.event_results_filtered(
    include=lambda er: er.result is not None,
    exclude=lambda er: isinstance(er.result, str) and 'error' in er.result,
)

This example includes only non-None results while excluding any string results containing the substring "error".

Using Default Truthy Filters

Call event_results_filtered without arguments to automatically exclude failed, incomplete, or empty results:

# Returns only successful, non‑None results

truthy = await event.event_results_filtered()

This uses the internal _event_result_is_truthy helper to retain only valid completed results.

Default Behaviors and Backward Compatibility

When you omit custom predicates, bubus applies sensible defaults:

• EventBus.expect: Default include returns True for all events; default exclude returns False for all events, effectively matching the first emitted event of the specified type.
• event_results_filtered: Default include uses _event_result_is_truthy, which filters out incomplete results, None values, exceptions, and forwarded events.

The legacy predicate parameter remains functional in both APIs. It is automatically combined with any explicit include callable, ensuring existing code continues to work while allowing migration to the newer dual-predicate interface.

Summary

• bubus provides predicate-based filtering at two stages: EventBus.expect for incoming events and BaseEvent.event_results_filtered for handler results.
• Both APIs accept include and exclude callables; a match requires include(event) == True and exclude(event) == False.
• The default filter for event_results_filtered uses _event_result_is_truthy in bubus/models.py to retain only successful, completed results.
• The legacy predicate argument is merged into include for backward compatibility, as implemented in bubus/service.py (lines 630-662).
• All predicates operate on the actual event or EventResult objects, enabling type-safe filtering based on any attribute or state.

Frequently Asked Questions

What is the difference between EventBus.expect and event_results_filtered?

EventBus.expect filters incoming events before they reach your code, waiting for a specific event type that matches your criteria. In contrast, event_results_filtered operates after event handlers execute, filtering the EventResult objects produced by those handlers. The former is located in bubus/service.py and works with live event streams, while the latter is defined in bubus/models.py and processes completed computations.

Can I use both include and exclude predicates together?

Yes. When you supply both parameters, bubus requires that the include callable returns True AND the exclude callable returns False for the same event or result. This allows precise filtering, such as including all non-None results while excluding specific error strings.

Is the predicate parameter deprecated?

While not formally deprecated, predicate is maintained for backward compatibility. According to the implementation in bubus/service.py, any predicate you provide is merged into the include callable using a lambda wrapper. New code should use include directly for clarity, but existing predicate usage remains fully functional.

What results pass the default filter in event_results_filtered?

By default, only results that satisfy _event_result_is_truthy pass the filter. As defined in bubus/models.py (lines 61-72), this includes only EventResult instances that are completed, not wrapped in exceptions, not forwarded events, and contain a non-None result value. Failed or pending results are automatically excluded.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}