agent-lab/src/agent_lab/application/events/registry.py

from __future__ import annotations

from collections.abc import Iterable
from copy import deepcopy

from agent_lab.application.events.models import EventDefinition


class EventRegistry:
    def __init__(self, definitions: Iterable[EventDefinition] = ()) -> None:
        self._definitions: dict[str, EventDefinition] = {}
        for definition in definitions:
            self.register(definition)

    def register(self, definition: EventDefinition) -> None:
        if definition.name in self._definitions:
            raise ValueError(f"duplicate event definition: {definition.name}")
        self._definitions[definition.name] = definition

    def definition(self, name: str) -> EventDefinition | None:
        return self._definitions.get(name)

    def catalog(self, enabled_names: Iterable[str] | None = None) -> list[dict]:
        enabled = set(enabled_names) if enabled_names is not None else None
        return [
            {
                "name": definition.name,
                "description": definition.description,
                "parameters": deepcopy(definition.parameters),
            }
            for definition in self._definitions.values()
            if enabled is None or definition.name in enabled
        ]

    def tool_schema(self, name: str) -> dict | None:
        definition = self.definition(name)
        if definition is None:
            return None
        return {
            "type": "function",
            "function": {
                "name": definition.name,
                "description": definition.description,
                "parameters": deepcopy(definition.parameters),
            },
        }