soulfire.plugins.ReflectivePlugin¶

class soulfire.plugins.ReflectivePlugin(descriptor: soulfire.plugin_api_pb2.PluginApiDescriptor, descriptor_set: bytes, client: connectrpc.client.ConnectClient, event_client: soulfire.plugin_api_connect.PluginApiServiceClient | None = None)[source]¶

Bases: _ReflectivePluginBase

call(service_name: str, method_name: str, value: dict[str, Any], *, headers: dict[str, str] | None = None, timeout_ms: int | None = None) → effect_py.Effect[ReflectiveMessage, soulfire.errors.SoulFireOperationError][source]¶

start_task(tasks: soulfire.tasks.SoulFireTasks, input_type_url: str, value: dict[str, Any], **options: Unpack[soulfire.tasks.TaskStartOptions]) → effect_py.Effect[soulfire.tasks.SoulFireTask[google.protobuf.message.Message], soulfire.errors.SoulFireOperationError][source]¶

stream(service_name: str, method_name: str, value: dict[str, Any], *, headers: dict[str, str] | None = None, timeout_ms: int | None = None) → soulfire.streams.Stream[ReflectiveMessage, soulfire.errors.SoulFireOperationError][source]¶

events(*, 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) → soulfire.streams.Stream[ReflectivePluginEvent, soulfire.errors.SoulFireOperationError][source]¶