soulfire.bot.SoulFireBot

class soulfire.bot.SoulFireBot(instance_id: str, bot_id: str, bot_client: soulfire.bot_connect.BotServiceClient, live_client: soulfire.bot_live_connect.BotLiveServiceClient, task_client: soulfire.task_connect.BotTaskServiceClient | None = None, pathfinder_client: soulfire.pathfinding_connect.PathfinderServiceClient | None = None, chat_client: soulfire.chat_connect.ChatServiceClient | None = None, inventory_client: soulfire.inventory_connect.InventoryServiceClient | None = None, recipe_client: soulfire.recipe_connect.RecipeServiceClient | None = None, registry_client: soulfire.registry_connect.RegistryServiceClient | None = None, world_client: soulfire.world_connect.WorldServiceClient | None = None, protocol_client: soulfire.protocol_connect.BotProtocolServiceClient | None = None)[source]

One bot’s lifecycle, actions, live state, and server tasks.

Obtain this handle from soulfire.SoulFire.create_bot(), soulfire.SoulFireInstance.get_or_create_bot(), or soulfire.SoulFireInstance.bot(). Direct construction requires RPC clients and is intended for transport integration.

Use connect() before reading state. The scope owns observation and stops a bot that this operation started. Task helpers return lazy Effects or Streams. A method call alone does not execute a bot operation.

Actions report rejection through the Effect error channel. Use tasks for server jobs and collect() for a collection workflow that waits for completion and cancels unfinished work on interruption.

instance_id
id
property state: soulfire.session.BotSessionState

Latest state from the session attached by connect().

Before connection or after scope cleanup, this property returns an empty state. It performs no network request. State can lag behind the server; use a session predicate to wait for a required update.

connect(*, ready_timeout: float = 30.0, timeout_ms: int | None = None) → effect_py.Effect[None, soulfire.errors.SoulFireOperationError, effect_py.Scope][source]

Start the bot if necessary and wait for its initial player snapshot.

An existing attached session makes this operation a no-op. Otherwise, it reads the bot’s desired state, starts a stopped bot, and attaches observation. The scope closes observation and stops the bot only if this operation started it. A previously running bot keeps its running state.

Parameters:

  • ready_timeout – Positive, finite deadline in seconds for startup and the initial player snapshot. Defaults to 30.

  • timeout_ms – Per-RPC timeout in milliseconds. This is separate from the overall readiness deadline.

Returns:

An Effect with no result value. Requires Scope.

Notes

An invalid deadline fails with SoulFireValidationError. Readiness expiration fails with SoulFireTimeoutError. RPC failures also use the Effect error channel. start() alone does not attach observation.

collect(target: str | collections.abc.Iterable[str], *, count: int = 1, search_radius: int = 32, avoid_submerged_targets: bool = False, require_line_of_sight: bool = False, options: soulfire.bot_live_pb2.PathfindOptions | None = None, timeout_ms: int | None = None) → effect_py.Effect[soulfire.task_pb2.CollectBlocksTaskResult, soulfire.errors.SoulFireOperationError][source]

Collect matching blocks and wait for the typed task result.

This helper owns the collection task. On interruption or failure before task completion, scope cleanup requests cancellation of unfinished server work. For a task handle with explicit ownership, use SoulFireTasks.collect_blocks() through tasks.

Parameters:

  • target – A block ID, block tag prefixed with #, or iterable of selectors. For example, "minecraft:oak_log" or "#minecraft:logs".

  • count – Number of blocks to collect. Defaults to 1.

  • search_radius – Search distance in blocks. Defaults to 32, at most 64.

  • avoid_submerged_targets – Skip targets covered by fluid up to the bot’s height.

  • require_line_of_sight – Restrict selection to visible blocks.

  • options – Pathfinding configuration, or server defaults when omitted.

  • timeout_ms – RPC timeout in milliseconds. This does not set a task deadline.

Returns:

An Effect that produces CollectBlocksTaskResult on completion. Non-successful task status fails through the Effect error channel.

Examples

Inside an Effect workflow:

result = yield from bot.collect("#minecraft:logs", count=16)
yield from sync(lambda: print(result))
property tasks: soulfire.tasks.SoulFireTasks

Durable server jobs for this bot.

Start methods return a task handle after acceptance. run_* methods return event streams with cancellation tied to the stream by default.

property pathfinder: soulfire.pathfinding.SoulFirePathfinder
property chat: soulfire.semantic.SoulFireChat
property inventory: soulfire.semantic.SoulFireInventory
property recipes: soulfire.semantic.SoulFireRecipes
property registry: soulfire.semantic.SoulFireRegistry
property world: soulfire.semantic.SoulFireWorld
property camera: soulfire.camera.SoulFireCamera
property protocol: soulfire.protocol.SoulFireProtocol
start(*, timeout_ms: int | None = None) → effect_py.Effect[soulfire.bot_pb2.BotStatus, soulfire.errors.SoulFireOperationError][source]

Set the bot’s desired state to running and return its status.

This operation does not wait for a player snapshot, attach observation, or stop the bot at scope exit. Use connect() for those lifecycle guarantees. timeout_ms controls the RPC timeout in milliseconds.

stop(*, timeout_ms: int | None = None) → effect_py.Effect[soulfire.bot_pb2.BotStatus, soulfire.errors.SoulFireOperationError][source]

Set the bot’s desired state to stopped and return its status.

This is an explicit server state change. It does not remove the account or instance. timeout_ms controls the RPC timeout in milliseconds.

restart(*, timeout_ms: int | None = None) → effect_py.Effect[soulfire.bot_pb2.BotStatus, soulfire.errors.SoulFireOperationError][source]
status(*, timeout_ms: int | None = None) → effect_py.Effect[soulfire.bot_pb2.BotStatus, soulfire.errors.SoulFireOperationError][source]
info(*, timeout_ms: int | None = None) → effect_py.Effect[soulfire.bot_pb2.BotInfoResponse, soulfire.errors.SoulFireOperationError][source]
wait_for_online(*, timeout_ms: int | None = None) → effect_py.Effect[soulfire.bot_pb2.BotStatus, soulfire.errors.SoulFireOperationError][source]

Wait for live state or a snapshot and return the latest bot status.

This operation does not start the bot or attach a session to state. An event stream that ends before readiness fails through the Effect error channel. timeout_ms controls RPC timeout in milliseconds.

events(event_filter: soulfire.bot_live_pb2.BotEventFilter | None = None, *, timeout_ms: int | None = None) → soulfire.streams.Stream[soulfire.bot_live_pb2.BotEvent, soulfire.errors.SoulFireOperationError][source]

Observe filtered bot events through a lazy Stream.

With no custom filter or timeout, an attached session supplies the events. Otherwise, consumption opens a separate RPC stream. The default filter includes state deltas, chat, lifecycle, inventory, damage, resource packs, and titles. Consuming events alone does not attach state.

Parameters:

  • event_filter – Explicit event selection, or the SDK default when omitted.

  • timeout_ms – RPC timeout in milliseconds.

Returns:

A Stream that produces BotEvent values. Interruption closes the subscription; it does not stop the bot.

observe(options: soulfire.session.BotSessionOptions | None = None, *, timeout_ms: int | None = None, ready_timeout: float | None = None) → effect_py.Effect[soulfire.session.BotSession, soulfire.errors.SoulFireOperationError, effect_py.Scope][source]

Open a scoped session that maintains state from bot events.

With no custom options, reuse the session attached by connect(), if present. Otherwise, return a separate session. The scope owns its subscription. This operation does not start a stopped bot or attach the new session to state; read the returned session’s state instead.

Parameters:

  • options – Session filters, buffering, and resumption configuration.

  • timeout_ms – RPC timeout in milliseconds.

  • ready_timeout – Session readiness wait in seconds. None uses the session default.

Returns:

An Effect that produces BotSession. Requires Scope.

send_chat(message: str, *, timeout_ms: int | None = None) → effect_py.Effect[soulfire.bot_live_pb2.BotActionResult, soulfire.errors.SoulFireOperationError][source]
get_block(position: soulfire.common_pb2.BlockPosition, *, timeout_ms: int | None = None) → effect_py.Effect[soulfire.bot_live_pb2.GetBlockResponse, soulfire.errors.SoulFireOperationError][source]
find_blocks(block_ids: collections.abc.Iterable[str], *, max_distance: int, max_count: int, timeout_ms: int | None = None) → effect_py.Effect[soulfire.bot_live_pb2.FindBlocksResponse, soulfire.errors.SoulFireOperationError][source]
list_nearby_entities(radius: float, *, entity_types: collections.abc.Iterable[str] = (), include_players: bool = True, timeout_ms: int | None = None) → effect_py.Effect[soulfire.bot_live_pb2.ListNearbyEntitiesResponse, soulfire.errors.SoulFireOperationError][source]
dig_block(position: soulfire.common_pb2.BlockPosition, *, cancel: bool = False, timeout_ms: int | None = None) → effect_py.Effect[soulfire.bot_live_pb2.BotActionResult, soulfire.errors.SoulFireOperationError][source]
place_block(against: soulfire.common_pb2.BlockPosition, face: soulfire.bot_live_pb2.BlockFace, hand: soulfire.bot_live_pb2.Hand = HAND_MAIN, *, timeout_ms: int | None = None) → effect_py.Effect[soulfire.bot_live_pb2.BotActionResult, soulfire.errors.SoulFireOperationError][source]
interact_block(position: soulfire.common_pb2.BlockPosition, face: soulfire.bot_live_pb2.BlockFace, hand: soulfire.bot_live_pb2.Hand = HAND_MAIN, *, sneaking: bool = False, timeout_ms: int | None = None) → effect_py.Effect[soulfire.bot_live_pb2.BotActionResult, soulfire.errors.SoulFireOperationError][source]
use_item(hand: soulfire.bot_live_pb2.Hand = HAND_MAIN, *, timeout_ms: int | None = None) → effect_py.Effect[soulfire.bot_live_pb2.BotActionResult, soulfire.errors.SoulFireOperationError][source]
release_item(*, timeout_ms: int | None = None) → effect_py.Effect[soulfire.bot_live_pb2.BotActionResult, soulfire.errors.SoulFireOperationError][source]
attack_entity(entity_id: int, *, sprinting: bool = False, timeout_ms: int | None = None) → effect_py.Effect[soulfire.bot_live_pb2.BotActionResult, soulfire.errors.SoulFireOperationError][source]
interact_entity(entity_id: int, *, hand: soulfire.bot_live_pb2.Hand = HAND_MAIN, sneaking: bool = False, timeout_ms: int | None = None) → effect_py.Effect[soulfire.bot_live_pb2.BotActionResult, soulfire.errors.SoulFireOperationError][source]
swing_arm(hand: soulfire.bot_live_pb2.Hand = HAND_MAIN, *, timeout_ms: int | None = None) → effect_py.Effect[soulfire.bot_live_pb2.BotActionResult, soulfire.errors.SoulFireOperationError][source]
respawn(*, timeout_ms: int | None = None) → effect_py.Effect[soulfire.bot_live_pb2.BotActionResult, soulfire.errors.SoulFireOperationError][source]
sleep(bed: soulfire.common_pb2.BlockPosition, hand: soulfire.bot_live_pb2.Hand = HAND_MAIN, *, timeout_ms: int | None = None) → effect_py.Effect[soulfire.bot_live_pb2.BotActionResult, soulfire.errors.SoulFireOperationError][source]
wake(*, timeout_ms: int | None = None) → effect_py.Effect[soulfire.bot_live_pb2.BotActionResult, soulfire.errors.SoulFireOperationError][source]
mount(entity_id: int, hand: soulfire.bot_live_pb2.Hand = HAND_MAIN, *, timeout_ms: int | None = None) → effect_py.Effect[soulfire.bot_live_pb2.MountEntityResponse, soulfire.errors.SoulFireOperationError][source]
dismount(*, timeout_ms: int | None = None) → effect_py.Effect[soulfire.bot_live_pb2.BotActionResult, soulfire.errors.SoulFireOperationError][source]
set_vehicle_control(*, forward: bool | None = None, backward: bool | None = None, left: bool | None = None, right: bool | None = None, jump: bool | None = None, sneak: bool | None = None, sprint: bool | None = None, yaw: float | None = None, pitch: float | None = None, timeout_ms: int | None = None) → effect_py.Effect[soulfire.bot_live_pb2.SetVehicleControlResponse, soulfire.errors.SoulFireOperationError][source]
update_sign(position: soulfire.common_pb2.BlockPosition, lines: collections.abc.Iterable[str], *, front_text: bool = True, timeout_ms: int | None = None) → effect_py.Effect[soulfire.bot_live_pb2.BotActionResult, soulfire.errors.SoulFireOperationError][source]
write_book(inventory_slot: int, pages: collections.abc.Iterable[str], *, title: str | None = None, timeout_ms: int | None = None) → effect_py.Effect[soulfire.bot_live_pb2.BotActionResult, soulfire.errors.SoulFireOperationError][source]
respond_resource_pack(pack_id: str, response: soulfire.bot_live_pb2.ResourcePackResponse, *, timeout_ms: int | None = None) → effect_py.Effect[soulfire.bot_live_pb2.BotActionResult, soulfire.errors.SoulFireOperationError][source]
set_flying(flying: bool, *, timeout_ms: int | None = None) → effect_py.Effect[soulfire.bot_live_pb2.BotActionResult, soulfire.errors.SoulFireOperationError][source]
start_elytra_flight(*, timeout_ms: int | None = None) → effect_py.Effect[soulfire.bot_live_pb2.BotActionResult, soulfire.errors.SoulFireOperationError][source]
set_creative_slot(slot: int, item_id: str | None = None, *, count: int = 1, timeout_ms: int | None = None) → effect_py.Effect[soulfire.bot_live_pb2.BotActionResult, soulfire.errors.SoulFireOperationError][source]
wait_for_chunks(radius_chunks: int = 0, *, wait_timeout_ms: int = 0, timeout_ms: int | None = None) → effect_py.Effect[soulfire.bot_live_pb2.WaitForChunksResponse, soulfire.errors.SoulFireOperationError][source]
go_to(goal: soulfire.bot_live_pb2.PathfindGoal, options: soulfire.bot_live_pb2.PathfindOptions | None = None, *, timeout_ms: int | None = None) → soulfire.streams.Stream[soulfire.bot_live_pb2.PathfindProgress, soulfire.errors.SoulFireOperationError][source]
stop_pathfinding(*, timeout_ms: int | None = None) → effect_py.Effect[None, soulfire.errors.SoulFireOperationError][source]
inventory_state(*, timeout_ms: int | None = None) → effect_py.Effect[soulfire.bot_pb2.BotInventoryStateResponse, soulfire.errors.SoulFireOperationError][source]
click_inventory(slot: int, click_type: soulfire.bot_pb2.ClickType = LEFT_CLICK, *, hotbar_slot: int = 0, timeout_ms: int | None = None) → effect_py.Effect[None, soulfire.errors.SoulFireOperationError][source]
transfer_inventory_slot(slot: int, *, timeout_ms: int | None = None) → effect_py.Effect[None, soulfire.errors.SoulFireOperationError][source]
drop_inventory_slot(slot: int, *, all: bool = True, timeout_ms: int | None = None) → effect_py.Effect[None, soulfire.errors.SoulFireOperationError][source]
move_inventory_stack(from_slot: int, to_slot: int, *, timeout_ms: int | None = None) → effect_py.Effect[None, soulfire.errors.SoulFireOperationError][source]
select_hotbar(slot: int, *, timeout_ms: int | None = None) → effect_py.Effect[None, soulfire.errors.SoulFireOperationError][source]
set_movement(*, forward: bool | None = None, backward: bool | None = None, left: bool | None = None, right: bool | None = None, jump: bool | None = None, sneak: bool | None = None, sprint: bool | None = None, timeout_ms: int | None = None) → effect_py.Effect[None, soulfire.errors.SoulFireOperationError][source]
reset_movement(*, timeout_ms: int | None = None) → effect_py.Effect[None, soulfire.errors.SoulFireOperationError][source]
look(yaw: float, pitch: float, *, timeout_ms: int | None = None) → effect_py.Effect[None, soulfire.errors.SoulFireOperationError][source]
open_inventory(*, timeout_ms: int | None = None) → effect_py.Effect[None, soulfire.errors.SoulFireOperationError][source]
close_container(*, timeout_ms: int | None = None) → effect_py.Effect[None, soulfire.errors.SoulFireOperationError][source]
click_container_button(button_id: int, *, timeout_ms: int | None = None) → effect_py.Effect[None, soulfire.errors.SoulFireOperationError][source]
acquire_control(*, ttl_seconds: int = 30, timeout_ms: int | None = None) → effect_py.Effect[SoulFireBotControlLease, soulfire.errors.SoulFireOperationError, effect_py.Scope][source]

Acquire exclusive action control and release it at scope exit.

Action requests from this handle include the lease token. Renew the lease before expiry for longer workflows; renewal is not automatic. Acquiring another lease on the same handle fails while a token is active.

Parameters:

  • ttl_seconds – Lease lifetime in seconds. Defaults to 30.

  • timeout_ms – RPC timeout in milliseconds.

Returns:

An Effect that produces a control lease. Requires Scope.

See also

SoulFireBotControlLease.renew() for manual renewal.

renew_control(lease: soulfire.bot_live_pb2.BotControlLease, ttl_seconds: int, timeout_ms: int | None) → effect_py.Effect[soulfire.bot_live_pb2.BotControlLease, soulfire.errors.SoulFireOperationError][source]
release_control(lease: soulfire.bot_live_pb2.BotControlLease, timeout_ms: int | None) → effect_py.Effect[None, soulfire.errors.SoulFireOperationError][source]