How to Merge Dict Results from Multiple Handlers in bubus

Use await event.event_results_flat_dict() to aggregate dictionary outputs from all event handlers, with built-in conflict detection and optional filtering.

When building event-driven applications with bubus, you often need to merge dict results from multiple handlers in bubus into a single consolidated mapping. The library provides a specialized method on BaseEvent that automatically collects, filters, and merges dictionary return values from every handler that processed the event.

Understanding the event_results_flat_dict Method

The core functionality resides in BaseEvent.event_results_flat_dict within [bubus/models.py](https://github.com/browser-use/bubus/blob/main/bubus/models.py#L600-L627). This asynchronous method assumes all handlers return dictionaries and merges them using sequential dict.update operations.

Method Signature and Parameters

async def event_results_flat_dict(
    self,
    timeout: float | None = None,
    include: EventResultFilter = _event_result_is_truthy,
    raise_if_any: bool = True,
    raise_if_none: bool = False,
    raise_if_conflicts: bool = True,
) -> dict[str, Any]:

• timeout: Maximum seconds to wait for handlers to complete.
• include: Callable filter to determine which EventResult objects participate in the merge.
• raise_if_any: Raise exception if any handler returns an error.
• raise_if_none: Raise exception if no handlers return valid results.
• raise_if_conflicts: Raise ValueError if dictionaries contain overlapping keys (default behavior).

Implementing Dictionary Result Merging

To merge dict results from multiple handlers in bubus, define an event with a dictionary result type, create handlers that return mappings, and invoke the aggregation method after dispatch.

Define an Event with Dict Result Type

from bubus import EventBus, BaseEvent

class GatherInfoEvent(BaseEvent[dict]):
    """Event that collects information from multiple services."""
    pass

Create Handlers Returning Dictionaries

def handler_user_data(event: GatherInfoEvent) -> dict:
    return {"user_id": 42, "name": "Alice"}

def handler_contact_info(event: GatherInfoEvent) -> dict:
    return {"email": "alice@example.com", "age": 30}

Register and Dispatch

bus = EventBus(name="mybus")
bus.on(GatherInfoEvent, handler_user_data)
bus.on(GatherInfoEvent, handler_contact_info)

async def main():
    event = await bus.dispatch(GatherInfoEvent())
    merged = await event.event_results_flat_dict()
    print(merged)
    # {'user_id': 42, 'name': 'Alice', 'email': 'alice@example.com', 'age': 30}

Handling Key Conflicts in Merged Results

By default, event_results_flat_dict raises a ValueError when two handlers return dictionaries containing the same key. This prevents accidental data loss when merging handler results.

Detecting Conflicts

def handler_conflict(event: GatherInfoEvent) -> dict:
    # Overlaps with handler_contact_info's "email" key

    return {"email": "alice@work.com", "department": "Engineering"}

bus.on(GatherInfoEvent, handler_conflict)

async def main():
    event = await bus.dispatch(GatherInfoEvent())
    try:
        merged = await event.event_results_flat_dict()
    except ValueError as exc:
        print("Key conflict detected:", exc)

Allowing Silent Overwrites

To implement a last-handler-wins strategy when you merge dict results from multiple handlers in bubus, set raise_if_conflicts=False:

async def main():
    event = await bus.dispatch(GatherInfoEvent())
    merged = await event.event_results_flat_dict(raise_if_conflicts=False)
    print(merged)
    # {'user_id': 42, 'name': 'Alice', 'email': 'alice@work.com', 'age': 30, 'department': 'Engineering'}

Filtering Which Handler Results to Include

The include parameter accepts any callable matching EventResultFilter to determine which results participate in the merge. This is useful when some handlers return empty dictionaries or None.

def only_non_empty(event_result):
    """Filter out empty dicts."""
    return isinstance(event_result.result, dict) and bool(event_result.result)

async def main():
    event = await bus.dispatch(GatherInfoEvent())
    merged = await event.event_results_flat_dict(include=only_non_empty)

Summary

• Use event_results_flat_dict in bubus/models.py to automatically merge dict results from multiple handlers in bubus.
• The method aggregates dictionaries using sequential dict.update from first to last handler.
• Enable raise_if_conflicts=True (default) to detect key collisions, or disable it for last-handler-wins behavior.
• Filter results using the include parameter to exclude empty or invalid dictionaries.
• This approach eliminates manual iteration over event_results_by_handler_name when all handlers return mappings.

Frequently Asked Questions

What happens if a handler returns None instead of a dict?

Results that are not dictionaries are automatically excluded from the merge. The default include filter (_event_result_is_truthy) filters out None values, empty dicts, and error results. Only valid dictionary objects participate in the final merge.

Can I merge results from specific handlers only?

Yes. While event_results_flat_dict processes all handlers by default, you can use the include parameter to filter by handler name, handler ID, or any other attribute of the EventResult object. For complete control, manually inspect event.event_results_by_handler_name and merge specific entries.

Does the merge order depend on handler registration order?

Yes. The dictionaries are merged in the order handlers were registered with the EventBus. The first registered handler's dict is updated with the second handler's dict, and so on. This means later handlers can overwrite earlier values when raise_if_conflicts=False.

What exception is raised when conflicting keys are detected?

When raise_if_conflicts=True (the default) and two handlers return dictionaries containing the same key, event_results_flat_dict raises a ValueError with a descriptive message indicating which handler caused the conflict and which key was duplicated.

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