跳转至

registry

feishu.integrations.registry

GatewayIntegration

Bases: Protocol

An external system that contributes routes to a Feishu gateway.

源代码位于： feishu/integrations/registry.py
Python
35
36
37
38
class GatewayIntegration(Protocol):
    r"""An external system that contributes routes to a Feishu gateway."""

    def routes(self, context: GatewayContext) -> Sequence[BaseRoute]: ...

IntegrationRegistry

Registers bundled tool and gateway integrations explicitly.

源代码位于： feishu/integrations/registry.py
Python
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
class IntegrationRegistry:
    r"""Registers bundled tool and gateway integrations explicitly."""

    def __init__(self) -> None:
        self._tool_bundles: dict[str, type[Any]] = {}
        self._gateway_factories: dict[str, GatewayIntegrationFactory] = {}

    @property
    def tool_bundle_names(self) -> tuple[str, ...]:
        return tuple(self._tool_bundles)

    @property
    def gateway_names(self) -> tuple[str, ...]:
        return tuple(self._gateway_factories)

    def register_tool_bundle(self, name: str, bundle: type[Any], *, override: bool = False) -> None:
        if name in self._tool_bundles and not override:
            raise ValueError(f"tool integration {name!r} is already registered")
        BUNDLES.register(bundle, name=name, override=override)
        self._tool_bundles[name] = bundle

    def register_gateway(self, name: str, factory: GatewayIntegrationFactory, *, override: bool = False) -> None:
        if name in self._gateway_factories and not override:
            raise ValueError(f"gateway integration {name!r} is already registered")
        self._gateway_factories[name] = factory

    def build_gateway(self, name: str, /, *args: Any, **kwargs: Any) -> GatewayIntegration:
        try:
            factory = self._gateway_factories[name]
        except KeyError as exc:
            available = ", ".join(self.gateway_names) or "<none>"
            raise ValueError(f"unknown gateway integration {name!r}; registered integrations: {available}") from exc
        return factory(*args, **kwargs)