ReflectivePlugin
Public Python members of ReflectivePlugin.
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.
call
@fn("ReflectivePlugin.call")
def call(
self,
service_name: str,
method_name: str,
value: dict[str, Any],
*,
headers: dict[str, str] | None = None,
timeout_ms: int | None = None,
) -> EffectGen[ReflectiveMessage, SoulFireOperationError]start_task
@fn("ReflectivePlugin.start_task")
def start_task(
self,
tasks: SoulFireTasks,
input_type_url: str,
value: dict[str, Any],
**options: Unpack[TaskStartOptions],
) -> EffectGen[SoulFireTask[Message], SoulFireOperationError]stream
def stream(
self,
service_name: str,
method_name: str,
value: dict[str, Any],
*,
headers: dict[str, str] | None = None,
timeout_ms: int | None = None,
) -> Stream[ReflectiveMessage, SoulFireOperationError]:
@genevents
def events(
self,
*,
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[ReflectivePluginEvent, SoulFireOperationError]:
@genHow is this page?
Last updated on