PluginCatalog
Public Python members of PluginCatalog.
Generated from the pinned SDK source. Use the matching installation. Each source link points to this baseline.
Methods decorated with @fn return effects. Their implementation signatures use EffectGen for the generator body. Compose them with yield from inside a scoped workflow.
require
def require[T](self, module: SoulFirePluginModule[T]) -> Tservice
def service[T](self, client_type: Callable[..., T]) -> Trefresh
@fn("PluginCatalog.refresh")
def refresh(
self, *, headers: dict[str, str] | None = None, timeout_ms: int | None = None
) -> EffectGen[tuple[PluginApiDescriptor, ...], SoulFireOperationError]descriptor_set
@fn("PluginCatalog.descriptor_set")
def descriptor_set(
self,
plugin_id: str,
*,
headers: dict[str, str] | None = None,
timeout_ms: int | None = None,
) -> EffectGen[bytes, SoulFireOperationError]reflective
@fn("PluginCatalog.reflective")
def reflective(
self,
plugin_id: str,
*,
headers: dict[str, str] | None = None,
timeout_ms: int | None = None,
) -> EffectGen[ReflectivePlugin, SoulFireOperationError]watch
def watch(
self, *, headers: dict[str, str] | None = None, timeout_ms: int | None = None
) -> Stream[PluginApiEvent, SoulFireOperationError]events
def events(
self,
*,
plugin_ids: tuple[str, ...] = (),
type_urls: tuple[str, ...] = (),
instance_id: str | None = None,
bot_id: str | None = None,
task_id: str | None = None,
after_sequence: int = 0,
headers: dict[str, str] | None = None,
timeout_ms: int | None = None,
) -> Stream[PluginEvent, SoulFireOperationError]typed_events
def typed_events[T: Message](
self,
plugin_id: str,
message_type: type[T],
*,
instance_id: str | None = None,
bot_id: str | None = None,
task_id: str | None = None,
after_sequence: int = 0,
headers: dict[str, str] | None = None,
timeout_ms: int | None = None,
) -> Stream[TypedPluginEvent[T], SoulFireOperationError]:
@genHow is this page?
Last updated on