mirrored 15 minutes ago
0
Linxin SongCoACT initialize (#292) b968155
# Copyright (c) 2023 - 2025, AG2ai, Inc., AG2ai open-source projects maintainers and core contributors
#
# SPDX-License-Identifier: Apache-2.0

from typing import Callable, TypeVar

from ..doc_utils import export_module
from .interoperable import Interoperable

__all__ = ["InteroperableRegistry", "register_interoperable_class"]

InteroperableClass = TypeVar("InteroperableClass", bound=type[Interoperable])


class InteroperableRegistry:
    def __init__(self) -> None:
        self._registry: dict[str, type[Interoperable]] = {}

    def register(self, short_name: str, cls: InteroperableClass) -> InteroperableClass:
        if short_name in self._registry:
            raise ValueError(f"Duplicate registration for {short_name}")

        self._registry[short_name] = cls

        return cls

    def get_short_names(self) -> list[str]:
        return sorted(self._registry.keys())

    def get_supported_types(self) -> list[str]:
        short_names = self.get_short_names()
        supported_types = [name for name in short_names if self._registry[name].get_unsupported_reason() is None]
        return supported_types

    def get_class(self, short_name: str) -> type[Interoperable]:
        return self._registry[short_name]

    @classmethod
    def get_instance(cls) -> "InteroperableRegistry":
        return _register


# global registry
_register = InteroperableRegistry()


# register decorator
@export_module("autogen.interop")
def register_interoperable_class(short_name: str) -> Callable[[InteroperableClass], InteroperableClass]:
    """Register an Interoperable class in the global registry.

    Returns:
        Callable[[InteroperableClass], InteroperableClass]: Decorator function

    Example:
        ```python
        @register_interoperable_class("myinterop")
        class MyInteroperability(Interoperable):
            def convert_tool(self, tool: Any) -> Tool:
                # implementation
                ...
        ```
    """

    def inner(cls: InteroperableClass) -> InteroperableClass:
        global _register
        return _register.register(short_name, cls)

    return inner