SDKReferenceSDK API

Python API

Core Python SDK operations and stream signatures.

Generated from the SoulFire SDK source. Each signature links to the same source revision.

Methods decorated with @fn return effects. Their source signatures use EffectGen for the generator body. Compose operations with yield from inside a scoped workflow. See Python setup for runtime and stream examples.

Connect and select bots

Compose operations with effect-py and keep connections inside a scope.

SoulFire

connect

@classmethod
@fn("SoulFire.connect")
def connect(
    cls,
    base_url: str,
    *,
    token: TokenProvider | None = None,
    timeout_ms: int | None = None,
    interceptors: Iterable[ClientInterceptor] = (),
    required_capabilities: Iterable[str] = (),
    required_plugins: Iterable[RequiredPlugin] = (),
) -> EffectGen[SoulFire, SoulFireOperationError, Scope]

install

@classmethod
@fn("SoulFire.install")
def install(
    cls,
    *,
    directory: str | os.PathLike[str] | None = None,
    version: str | None = None,
    java_args: Iterable[str] = (),
    port: int | None = None,
    startup_timeout: float = 120.0,
    on_log: Callable[[str], None] | None = None,
    timeout_ms: int | None = None,
    interceptors: Iterable[ClientInterceptor] = (),
) -> EffectGen[SoulFire, SoulFireOperationError, Scope]

instance

def instance(self, instance_id: str) -> SoulFireInstance

instances

@fn("SoulFire.instances")
def instances(
    self, *, headers: dict[str, str] | None = None, timeout_ms: int | None = None
) -> EffectGen[list[InstanceListResponse.Instance], SoulFireOperationError]

close

@fn("SoulFire.close")
def close(self) -> EffectGen[None]

SoulFireInstance

bot

def bot(self, bot_id: str) -> SoulFireBot

bots

@fn("SoulFireInstance.bots")
def bots(
    self, *, headers: dict[str, str] | None = None, timeout_ms: int | None = None
) -> EffectGen[list[BotListEntry], SoulFireOperationError]

events

def events(
    self,
    filter: InstanceEventFilter | None = None,
    *,
    bot_ids: Iterable[str] = (),
    headers: dict[str, str] | None = None,
    timeout_ms: int | None = None,
) -> Stream[InstanceEvent, SoulFireOperationError]

start

@fn("SoulFireInstance.start")
def start(
    self,
    *,
    bot_ids: Iterable[str] | None = None,
    count: int | None = None,
    headers: dict[str, str] | None = None,
    timeout_ms: int | None = None,
) -> EffectGen[list[BotStatus], SoulFireOperationError]

stop

@fn("SoulFireInstance.stop")
def stop(
    self,
    *,
    bot_ids: Iterable[str] | None = None,
    count: int | None = None,
    headers: dict[str, str] | None = None,
    timeout_ms: int | None = None,
) -> EffectGen[list[BotStatus], SoulFireOperationError]

Control a bot

Operations return effects. Streams expose effect-based consumers.

SoulFireBot

start

@fn("SoulFireBot.start")
def start(
    self, *, timeout_ms: int | None = None
) -> EffectGen[BotStatus, SoulFireOperationError]

stop

@fn("SoulFireBot.stop")
def stop(
    self, *, timeout_ms: int | None = None
) -> EffectGen[BotStatus, SoulFireOperationError]

status

@fn("SoulFireBot.status")
def status(
    self, *, timeout_ms: int | None = None
) -> EffectGen[BotStatus, SoulFireOperationError]

events

def events(
    self, event_filter: BotEventFilter | None = None, *, timeout_ms: int | None = None
) -> Stream[BotEvent, SoulFireOperationError]

observe

@fn("SoulFireBot.observe")
def observe(
    self, options: BotSessionOptions | None = None, *, timeout_ms: int | None = None
) -> EffectGen[BotSession, SoulFireOperationError, Scope]

wait_for_online

@fn("SoulFireBot.wait_for_online")
def wait_for_online(
    self, *, timeout_ms: int | None = None
) -> EffectGen[BotStatus, SoulFireOperationError]:
    @gen

SoulFireChat

send

@fn("SoulFireChat.send")
def send(
    self, message: str, *, idempotency_key: str | None = None, timeout_ms: int | None = None
) -> EffectGen[BotActionResult, SoulFireOperationError]

command

@fn("SoulFireChat.command")
def command(
    self, command: str, *, idempotency_key: str | None = None, timeout_ms: int | None = None
) -> EffectGen[BotActionResult, SoulFireOperationError]

watch

def watch(
    self,
    matcher: ChatMatcher,
    *,
    sources: Iterable[ChatSource] | None = None,
    timeout_ms: int | None = None,
) -> Stream[ChatMatch, SoulFireOperationError]:
    @gen

wait_for

@fn("SoulFireChat.wait_for")
def wait_for(
    self,
    matcher: ChatMatcher,
    *,
    sources: Iterable[ChatSource] | None = None,
    timeout_ms: int | None = None,
) -> EffectGen[ChatMatch, SoulFireOperationError]

Run durable tasks

Task handles survive SDK disconnects. Read their state or resume them by ID.

SoulFireTasks

go_to

@fn("SoulFireTasks.go_to")
def go_to(
    self,
    goal: PathfindGoal,
    *,
    options: PathfindOptions | None = None,
    conflict_policy: BotTaskConflictPolicy = BOT_TASK_CONFLICT_POLICY_UNSPECIFIED,
    reconnect_policy: BotTaskReconnectPolicy = BOT_TASK_RECONNECT_POLICY_UNSPECIFIED,
    priority: BotTaskPriority = BOT_TASK_PRIORITY_UNSPECIFIED,
    deadline: datetime | None = None,
    idempotency_key: str | None = None,
    headers: dict[str, str] | None = None,
    timeout_ms: int | None = None,
) -> EffectGen[SoulFireTask[GoToTaskResult], SoulFireOperationError]

follow_entity

@fn("SoulFireTasks.follow_entity")
def follow_entity(
    self,
    target: FollowEntityTarget,
    *,
    distance: float = 3,
    options: PathfindOptions | None = None,
    target_unavailable_timeout_seconds: int = 0,
    conflict_policy: BotTaskConflictPolicy = BOT_TASK_CONFLICT_POLICY_UNSPECIFIED,
    reconnect_policy: BotTaskReconnectPolicy = BOT_TASK_RECONNECT_POLICY_UNSPECIFIED,
    priority: BotTaskPriority = BOT_TASK_PRIORITY_UNSPECIFIED,
    deadline: datetime | None = None,
    idempotency_key: str | None = None,
    headers: dict[str, str] | None = None,
    timeout_ms: int | None = None,
) -> EffectGen[SoulFireTask[FollowEntityTaskResult], SoulFireOperationError]

collect_blocks

@fn("SoulFireTasks.collect_blocks")
def collect_blocks(
    self,
    block_ids: Iterable[str] = (),
    *,
    tags: Iterable[str] = (),
    count: int = 1,
    search_radius: int = 32,
    avoid_submerged_targets: bool = False,
    require_line_of_sight: bool = False,
    target_y_range: IntRange | None = None,
    options: PathfindOptions | None = None,
    conflict_policy: BotTaskConflictPolicy = BOT_TASK_CONFLICT_POLICY_UNSPECIFIED,
    reconnect_policy: BotTaskReconnectPolicy = BOT_TASK_RECONNECT_POLICY_UNSPECIFIED,
    priority: BotTaskPriority = BOT_TASK_PRIORITY_UNSPECIFIED,
    deadline: datetime | None = None,
    idempotency_key: str | None = None,
    headers: dict[str, str] | None = None,
    timeout_ms: int | None = None,
) -> EffectGen[SoulFireTask[CollectBlocksTaskResult], SoulFireOperationError]

stash

@fn("SoulFireTasks.stash")
def stash(
    self,
    container: BlockPosition,
    operations: Iterable[ContainerTransferSpec | ContainerTransferOperation],
    *,
    options: PathfindOptions | None = None,
    close_container: bool = True,
    conflict_policy: BotTaskConflictPolicy = BOT_TASK_CONFLICT_POLICY_UNSPECIFIED,
    reconnect_policy: BotTaskReconnectPolicy = BOT_TASK_RECONNECT_POLICY_UNSPECIFIED,
    priority: BotTaskPriority = BOT_TASK_PRIORITY_UNSPECIFIED,
    deadline: datetime | None = None,
    idempotency_key: str | None = None,
    headers: dict[str, str] | None = None,
    timeout_ms: int | None = None,
) -> EffectGen[SoulFireTask[ContainerTransferTaskResult], SoulFireOperationError]

get

@fn("SoulFireTasks.get")
def get[ResultT: Message](
    self,
    task_id: str,
    result_type: type[ResultT],
    *,
    headers: dict[str, str] | None = None,
    timeout_ms: int | None = None,
) -> EffectGen[SoulFireTask[ResultT], SoulFireOperationError]

list

@fn("SoulFireTasks.list")
def list(
    self,
    *,
    statuses: Iterable[BotTaskStatus] = (),
    include_terminal: bool = False,
    page_size: int = 100,
    headers: dict[str, str] | None = None,
    timeout_ms: int | None = None,
) -> EffectGen[list[BotTask], SoulFireOperationError]

watch

def watch(
    self,
    *,
    statuses: Iterable[BotTaskStatus] = (),
    after_sequence: int = 0,
    include_snapshot: bool = True,
    headers: dict[str, str] | None = None,
    timeout_ms: int | None = None,
) -> Stream[BotTaskEvent, SoulFireOperationError]

SoulFireTask

refresh

@fn("SoulFireTask.refresh")
def refresh(
    self, *, headers: dict[str, str] | None = None, timeout_ms: int | None = None
) -> EffectGen[BotTask, SoulFireOperationError]

events

def events(
    self,
    *,
    after_revision: int | None = None,
    headers: dict[str, str] | None = None,
    timeout_ms: int | None = None,
) -> Stream[BotTaskEvent, SoulFireOperationError]

wait

@fn("SoulFireTask.wait")
def wait(
    self, *, headers: dict[str, str] | None = None, timeout_ms: int | None = None
) -> EffectGen[BotTask, SoulFireOperationError]

cancel

@fn("SoulFireTask.cancel")
def cancel(
    self,
    reason: str = "",
    *,
    headers: dict[str, str] | None = None,
    timeout_ms: int | None = None,
) -> EffectGen[BotTask, SoulFireOperationError]

result

@fn("SoulFireTask.result")
def result(
    self, *, headers: dict[str, str] | None = None, timeout_ms: int | None = None
) -> EffectGen[ResultT, SoulFireOperationError]

How is this page?

Last updated on

TypeScript API

Core TypeScript SDK operations and stream signatures.

TypeScript options

Connection, session, and task options from the SDK source.

On this page

Connect and select bots SoulFire SoulFireInstance Control a bot SoulFireBot SoulFireChat Run durable tasks SoulFireTasks SoulFireTask