Skip to content

simulatecraft.minecraft

simulatecraft.minecraft

SimulateCraft Minecraft module.

Public API

from simulatecraft.minecraft import (
    MinecraftEnvironment,
    MinecraftBridge,
    MinecraftObservation,
    ALL_ACTIONS,
    # individual actions:
    Move, Jump, Sneak, LookAt,
    MineBlock, PlaceBlock, UseItem, ActivateBlock,
    EquipItem, DropItem, Craft,
    Chat, Whisper,
    NavigateTo, FollowEntity,
    Wait,
)

ActivateBlock

Bases: Action

Right-click a block to open/use it (chest, furnace, door, lever, etc.).

Chat

Bases: Action

Send a public message in Minecraft chat.

Craft

Bases: Action

Craft an item by name (recipe looked up on the bot side).

DropItem

Bases: Action

Drop one or more of an item from inventory.

EquipItem

Bases: Action

Move an item to the bot's hand or armour slot.

FollowEntity

Bases: Action

Follow a named player or mob until the next action.

Jump

Bases: Action

Jump (optionally while moving).

LookAt

Bases: Action

Turn to face a target — block coordinates or entity name.

MineBlock

Bases: Action

Dig/break a block at a given position (or the block the bot is looking at).

Move

Bases: Action

Walk one step in a cardinal direction.

NavigateTo

Bases: Action

High-level pathfind to a position. Mineflayer pathfinder handles obstacles.

PlaceBlock

Bases: Action

Place a block from inventory at a given position.

Sneak

Bases: Action

Toggle sneaking on or off.

UseItem

Bases: Action

Right-click / use the currently equipped item (optionally on a target block).

Wait

Bases: Action

Do nothing for a number of ticks. Useful when waiting for events.

Whisper

Bases: Action

Send a private /msg to another player.

BridgeError

Bases: Exception

Raised when the bot process crashes or returns an error response.

MinecraftBridge 

MinecraftBridge(*, host: str = 'localhost', minecraft_port: int = 25565, username: str = 'SimBot', password: str = '', version: str | None = None, ipc_port: int = _DEFAULT_IPC_PORT, bot_script: str | Path | None = None, node_executable: str = 'node', connect_timeout: float = 30.0, request_timeout: float = 45.0, auth: str = 'offline')

Manages the Node.js Mineflayer subprocess and the JSON-RPC socket.

Source code in src/simulatecraft/minecraft/connection.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def __init__(
    self,
    *,
    host: str = "localhost",
    minecraft_port: int = 25565,
    username: str = "SimBot",
    password: str = "",
    version: str | None = None,
    ipc_port: int = _DEFAULT_IPC_PORT,
    bot_script: str | Path | None = None,
    node_executable: str = "node",
    connect_timeout: float = 30.0,
    request_timeout: float = 45.0,
    auth: str = "offline",
) -> None:
    self.host = host
    self.minecraft_port = minecraft_port
    self.username = username
    self.password = password
    self.version = version
    self.ipc_port = ipc_port
    self.bot_script = Path(bot_script) if bot_script else _DEFAULT_BOT_SCRIPT
    self.node_executable = node_executable
    self.connect_timeout = connect_timeout
    self.request_timeout = request_timeout
    self.auth = auth

    self._process: subprocess.Popen | None = None
    self._reader: asyncio.StreamReader | None = None
    self._writer: asyncio.StreamWriter | None = None
    self._pending: dict[str, asyncio.Future[Any]] = {}
    self._event_handlers: dict[str, list[Any]] = {}
    self._read_task: asyncio.Task | None = None
    self._connected = False

connect async 

connect() -> None

Spawn the Node bot and wait until it signals it has joined the server.

Source code in src/simulatecraft/minecraft/connection.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
async def connect(self) -> None:
    """Spawn the Node bot and wait until it signals it has joined the server."""
    if self._connected:
        return

    if not self.bot_script.exists():
        raise FileNotFoundError(
            f"Bot script not found: {self.bot_script}\n"
            "Run: cd src/simulatecraft/minecraft/bot && npm install"
        )

    env = {**os.environ, "IPC_PORT": str(self.ipc_port)}
    cmd = [
        self.node_executable,
        str(self.bot_script),
        "--host",
        self.host,
        "--port",
        str(self.minecraft_port),
        "--username",
        self.username,
        "--ipc-port",
        str(self.ipc_port),
        "--auth",
        self.auth,
    ]
    if self.password:
        cmd += ["--password", self.password]
    if self.version:
        cmd += ["--version", self.version]

    log.info("Spawning bot process: %s", " ".join(cmd))
    self._process = subprocess.Popen(  # noqa: S603
        cmd,
        env=env,
        cwd=str(self.bot_script.parent),
        stdout=None,
        stderr=None,
    )

    await self._tcp_connect()

    spawned: asyncio.Future[dict[str, Any]] = asyncio.get_running_loop().create_future()

    def _on_spawned(data: dict[str, Any]) -> None:
        if not spawned.done():
            spawned.set_result(data)

    def _on_failed(data: dict[str, Any]) -> None:
        if spawned.done():
            return
        reason = data.get("message") or data.get("reason") or "disconnected"
        spawned.set_exception(BridgeError(self._minecraft_connect_hint(str(reason))))

    self.on_event("bot.spawned", _on_spawned)
    self.on_event("bot.error", _on_failed)
    self.on_event("bot.disconnected", _on_failed)
    try:
        await asyncio.wait_for(spawned, timeout=self.connect_timeout)
    except TimeoutError as exc:
        await self.close()
        raise BridgeError(
            f"Bot did not spawn within {self.connect_timeout}s. "
            f"Is a Minecraft Java server running at {self.host}:{self.minecraft_port}?"
        ) from exc
    except BridgeError:
        await self.close()
        raise

    self._connected = True
    log.info("Bot '%s' spawned in Minecraft.", self.username)

close async 

close() -> None

Gracefully shut down the bot and the subprocess.

Source code in src/simulatecraft/minecraft/connection.py 
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
async def close(self) -> None:
    """Gracefully shut down the bot and the subprocess."""
    self._connected = False
    if self._read_task and not self._read_task.done():
        self._read_task.cancel()
    if self._writer:
        try:
            self._writer.close()
            await self._writer.wait_closed()
        except Exception:
            pass
    if self._process and self._process.poll() is None:
        self._process.terminate()
        try:
            self._process.wait(timeout=5)
        except subprocess.TimeoutExpired:
            self._process.kill()
    # Fail any pending requests
    for fut in self._pending.values():
        if not fut.done():
            fut.set_exception(BridgeError("bridge closed"))
    self._pending.clear()

call async 

call(method: str, **params: Any) -> Any

Send an RPC request and await its response.

Source code in src/simulatecraft/minecraft/connection.py 
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
async def call(self, method: str, **params: Any) -> Any:
    """Send an RPC request and await its response."""
    if self._writer is None:
        raise BridgeError("Bridge not connected. Call connect() first.")
    req_id = str(uuid.uuid4())
    loop = asyncio.get_event_loop()
    future: asyncio.Future[Any] = loop.create_future()
    self._pending[req_id] = future
    msg = json.dumps({"id": req_id, "method": method, "params": params}) + "\n"
    self._writer.write(msg.encode())
    await self._writer.drain()
    try:
        return await asyncio.wait_for(future, timeout=self.request_timeout)
    except TimeoutError as exc:
        self._pending.pop(req_id, None)
        raise BridgeError(f"RPC '{method}' timed out after {self.request_timeout}s") from exc

get_state async 

get_state() -> dict[str, Any]

Return a full world-state snapshot from the bot.

Source code in src/simulatecraft/minecraft/connection.py 
263
264
265
async def get_state(self) -> dict[str, Any]:
    """Return a full world-state snapshot from the bot."""
    return await self.call("get_state")

perform_action async 

perform_action(action: dict[str, Any]) -> dict[str, Any]

Execute one action dict (matches Action.model_dump()) on the bot.

Source code in src/simulatecraft/minecraft/connection.py 
267
268
269
async def perform_action(self, action: dict[str, Any]) -> dict[str, Any]:
    """Execute one action dict (matches Action.model_dump()) on the bot."""
    return await self.call("perform_action", action=action)

configure_presence async 

configure_presence(*, x: float | None = None, y: float | None = None, z: float | None = None, gamemode: str | None = None) -> dict[str, Any]

Teleport after spawn via chat (RCON is preferred for reliability).

Source code in src/simulatecraft/minecraft/connection.py 
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
async def configure_presence(
    self,
    *,
    x: float | None = None,
    y: float | None = None,
    z: float | None = None,
    gamemode: str | None = None,
) -> dict[str, Any]:
    """Teleport after spawn via chat (RCON is preferred for reliability)."""
    _ = gamemode  # agents never set gamemode; kept for call-site compatibility
    return await self.call(
        "configure_presence",
        x=x,
        y=y,
        z=z,
    )

AgentBotConfig 

AgentBotConfig(username: str, password: str = '', ipc_port: int = 25570, auth: str = 'offline', goal: str = '', spawn_x: float | None = None, spawn_y: float | None = None, spawn_z: float | None = None, persona: str = '')

Per-agent bot connection settings.

Source code in src/simulatecraft/minecraft/env.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
def __init__(
    self,
    username: str,
    password: str = "",
    ipc_port: int = 25570,
    auth: str = "offline",
    goal: str = "",
    spawn_x: float | None = None,
    spawn_y: float | None = None,
    spawn_z: float | None = None,
    persona: str = "",
) -> None:
    self.username = username
    self.password = password
    self.ipc_port = ipc_port
    self.auth = auth
    self.goal = goal
    self.spawn_x = spawn_x
    self.spawn_y = spawn_y
    self.spawn_z = spawn_z
    self.persona = persona

MinecraftEnvironment 

MinecraftEnvironment(*, server_host: str = 'localhost', server_port: int = 25565, version: str | None = None, bot_script: str | Path | None = None, node_executable: str = 'node', block_scan_radius: int = 6, entity_scan_radius: int = 16, chat_log_size: int = 20, connect_timeout: float = 30.0, request_timeout: float = 45.0)

Bases: Environment

Multi-agent Minecraft environment backed by Mineflayer bots.

Each registered agent maps to one bot subprocess. The environment queries each bot's state for observe() and dispatches actions back through the bridge in step().

Source code in src/simulatecraft/minecraft/env.py 
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def __init__(
    self,
    *,
    server_host: str = "localhost",
    server_port: int = 25565,
    version: str | None = None,
    bot_script: str | Path | None = None,
    node_executable: str = "node",
    block_scan_radius: int = 6,
    entity_scan_radius: int = 16,
    chat_log_size: int = 20,
    connect_timeout: float = 30.0,
    request_timeout: float = 45.0,
) -> None:
    super().__init__()
    self.server_host = server_host
    self.server_port = server_port
    self.version = version
    self.bot_script = bot_script
    self.node_executable = node_executable
    self.block_scan_radius = block_scan_radius
    self.entity_scan_radius = entity_scan_radius
    self.chat_log_size = chat_log_size
    self.connect_timeout = connect_timeout
    self.request_timeout = request_timeout

    # agent_id → bridge
    self._bridges: dict[str, MinecraftBridge] = {}
    # agent_id → config
    self._bot_configs: dict[str, AgentBotConfig] = {}
    # per-agent rolling chat log
    self._chat_logs: dict[str, list[ChatMessage]] = {}
    # per-agent last reward (set by step() for observe() to return)
    self._last_rewards: dict[str, float] = {}
    self._map_cache: dict[str, Any] | None = None
    self._map_size: int = 128
    self._map_origin: tuple[int, int] | None = None
    self._home_xz: tuple[int, int] | None = None
    self._map_pan_limit: int = 512

add_bot 

add_bot(agent_id: str, *, username: str | None = None, password: str = '', ipc_port: int | None = None, auth: str = 'offline', goal: str = '', spawn_x: float | None = None, spawn_y: float | None = None, spawn_z: float | None = None, persona: str = '') -> None

Register an agent and configure its bot.

Call before connect(), or use :meth:spawn_bot to add one at runtime. ipc_port defaults to the next free port starting at 25570.

Source code in src/simulatecraft/minecraft/env.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
def add_bot(
    self,
    agent_id: str,
    *,
    username: str | None = None,
    password: str = "",
    ipc_port: int | None = None,
    auth: str = "offline",
    goal: str = "",
    spawn_x: float | None = None,
    spawn_y: float | None = None,
    spawn_z: float | None = None,
    persona: str = "",
) -> None:
    """Register an agent and configure its bot.

    Call before ``connect()``, or use :meth:`spawn_bot` to add one at runtime.
    ``ipc_port`` defaults to the next free port starting at 25570.
    """
    if agent_id in self._bot_configs:
        raise ValueError(f"agent {agent_id!r} already registered")
    if ipc_port is None:
        ipc_port = self._next_ipc_port()
    self._bot_configs[agent_id] = AgentBotConfig(
        username=username or agent_id,
        password=password,
        ipc_port=ipc_port,
        auth=auth,
        goal=goal,
        spawn_x=spawn_x,
        spawn_y=spawn_y,
        spawn_z=spawn_z,
        persona=persona,
    )
    self._chat_logs[agent_id] = []
    self._last_rewards[agent_id] = 0.0
    self.register_agent(agent_id)

connect async 

connect() -> None

Spawn all bots and wait for them to join the server.

Source code in src/simulatecraft/minecraft/env.py 
174
175
176
177
178
179
180
async def connect(self) -> None:
    """Spawn all bots and wait for them to join the server."""
    tasks = [self._connect_one(aid) for aid in self._bot_configs]
    await asyncio.gather(*tasks)
    # Warm observation cache so the viewer and first decide() see real state.
    await self._fetch_all_states()
    await self._refresh_map()

spawn_bot async 

spawn_bot(agent_id: str, *, username: str | None = None, password: str = '', auth: str = 'offline', goal: str = '', spawn_x: float | None = None, spawn_y: float | None = None, spawn_z: float | None = None, persona: str = '') -> None

Register and connect a bot while the environment is already running.

Source code in src/simulatecraft/minecraft/env.py 
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
async def spawn_bot(
    self,
    agent_id: str,
    *,
    username: str | None = None,
    password: str = "",
    auth: str = "offline",
    goal: str = "",
    spawn_x: float | None = None,
    spawn_y: float | None = None,
    spawn_z: float | None = None,
    persona: str = "",
) -> None:
    """Register and connect a bot while the environment is already running."""
    self.add_bot(
        agent_id,
        username=username,
        password=password,
        auth=auth,
        goal=goal,
        spawn_x=spawn_x,
        spawn_y=spawn_y,
        spawn_z=spawn_z,
        persona=persona,
    )
    try:
        await self._connect_one(agent_id)
        await self._fetch_all_states()
    except Exception:
        await self.despawn_bot(agent_id)
        raise

despawn_bot async 

despawn_bot(agent_id: str) -> None

Disconnect one bot and forget its registration.

Source code in src/simulatecraft/minecraft/env.py 
248
249
250
251
252
253
254
255
256
257
258
259
260
async def despawn_bot(self, agent_id: str) -> None:
    """Disconnect one bot and forget its registration."""
    bridge = self._bridges.pop(agent_id, None)
    if bridge is not None:
        with contextlib.suppress(Exception):
            await bridge.close()
    self._bot_configs.pop(agent_id, None)
    self._chat_logs.pop(agent_id, None)
    self._last_rewards.pop(agent_id, None)
    cache = getattr(self, "_obs_cache", None)
    if isinstance(cache, dict):
        cache.pop(agent_id, None)
    self.unregister_agent(agent_id)

close async 

close() -> None

Disconnect all bots gracefully.

Source code in src/simulatecraft/minecraft/env.py 
288
289
290
291
async def close(self) -> None:
    """Disconnect all bots gracefully."""
    await asyncio.gather(*(b.close() for b in self._bridges.values()))
    self._bridges.clear()

prepare_tick async 

prepare_tick() -> None

Refresh bot observations before the runner asks each agent to decide.

Source code in src/simulatecraft/minecraft/env.py 
304
305
306
307
async def prepare_tick(self) -> None:
    """Refresh bot observations before the runner asks each agent to decide."""
    await self._fetch_all_states()
    await self._refresh_map()

observe 

observe(agent_id: str) -> MinecraftObservation

Return the latest cached observation for this agent.

Source code in src/simulatecraft/minecraft/env.py 
309
310
311
312
313
314
315
316
317
318
319
def observe(self, agent_id: str) -> MinecraftObservation:
    """Return the latest cached observation for this agent."""
    cached = getattr(self, "_obs_cache", {}).get(agent_id)
    if cached is not None:
        return cached
    cfg = self._bot_configs.get(agent_id, AgentBotConfig(username=agent_id))
    return MinecraftObservation(
        agent_id=agent_id,
        tick=self._tick_count,
        current_goal=cfg.goal,
    )

step async 

step(agent_id: str, action: Action) -> StepResult

Dispatch the action to the bot and wait for Mineflayer to finish it.

Source code in src/simulatecraft/minecraft/env.py 
321
322
323
324
325
326
327
328
329
async def step(self, agent_id: str, action: Action) -> StepResult:
    """Dispatch the action to the bot and wait for Mineflayer to finish it."""
    bridge = self._bridges.get(agent_id)
    if bridge is None:
        return StepResult(info={"error": f"no bridge for agent {agent_id}"})
    result = await self._execute_action(agent_id, bridge, action)
    reward = self._last_rewards.get(agent_id, 0.0)
    info = {"action": action.kind, **(result if isinstance(result, dict) else {})}
    return StepResult(reward=reward, info=info)

tick 

tick() -> None

Advance the environment clock. Observations refresh in prepare_tick().

Source code in src/simulatecraft/minecraft/env.py 
350
351
352
def tick(self) -> None:
    """Advance the environment clock. Observations refresh in prepare_tick()."""
    self._tick_count += 1

fetch_map async 

fetch_map(origin_x: int, origin_z: int, size: int | None = None) -> dict[str, Any]

Scan a top-down map tile for the viewer (also used by WS pan requests).

Source code in src/simulatecraft/minecraft/env.py 
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
async def fetch_map(
    self, origin_x: int, origin_z: int, size: int | None = None
) -> dict[str, Any]:
    """Scan a top-down map tile for the viewer (also used by WS pan requests)."""
    if not self._bridges:
        return {}
    tile = max(16, min(int(size or self._map_size), 128))
    ox = int(origin_x)
    oz = int(origin_z)
    if self._home_xz is not None:
        hx, hz = self._home_xz
        lim = self._map_pan_limit
        ox = max(hx - lim, min(hx + lim - tile, ox))
        oz = max(hz - lim, min(hz + lim - tile, oz))
    bridge = next(iter(self._bridges.values()))
    result = await bridge.get_map(ox, oz, tile)
    self._map_cache = result
    self._map_origin = (ox, oz)
    return result

snapshot 

snapshot() -> Snapshot

Top-down Minecraft map (surface blocks) plus agent markers in world XZ.

Source code in src/simulatecraft/minecraft/env.py 
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
def snapshot(self) -> Snapshot:
    """Top-down Minecraft map (surface blocks) plus agent markers in world XZ."""
    cache = getattr(self, "_obs_cache", {})
    agents: dict[str, dict[str, Any]] = {}
    for aid in self.agent_ids:
        cfg = self._bot_configs.get(aid)
        obs = cache.get(aid)
        if obs is not None:
            x, y, z = obs.position.x, obs.position.y, obs.position.z
            agents[aid] = {
                "position": [x, z],
                "position_3d": [x, y, z],
                "name": cfg.username if cfg else aid,
                "health": obs.stats.health,
                "food": obs.stats.food,
                "holding": obs.equipped_item,
                "goal": obs.current_goal,
                "biome": obs.biome,
                "yaw": obs.yaw,
                "persona": cfg.persona if cfg else "",
            }
        else:
            agents[aid] = {
                "position": [0.0, 0.0],
                "name": cfg.username if cfg else aid,
                "goal": cfg.goal if cfg else "",
                "persona": cfg.persona if cfg else "",
            }

    world_map = self._map_cache or {}
    width = int(world_map.get("width") or self._map_size)
    height = int(world_map.get("height") or self._map_size)
    origin_x = world_map.get("origin_x", 0)
    origin_z = world_map.get("origin_z", 0)
    if self._home_xz is not None:
        home = list(self._home_xz)
    else:
        home = [origin_x + width // 2, origin_z + height // 2]

    return Snapshot(
        tick=self._tick_count,
        agents=agents,
        world={
            "kind": "minecraft",
            "server": self.server_host,
            "view": "top-down-xz",
            "width": width,
            "height": height,
            "origin_xz": [origin_x, origin_z],
            "home_xz": home,
            "pan_limit": self._map_pan_limit,
            "tile_size": self._map_size,
            "map": world_map,
        },
    )

BotStats

Bases: BaseModel

Vital statistics of the bot.

ChatMessage

Bases: BaseModel

One line from the Minecraft chat log.

InventoryItem

Bases: BaseModel

One stack in the bot's inventory.

MinecraftObservation

Bases: Observation

Full structured state snapshot handed to the agent brain each tick.

All fields have sensible defaults so partial observations work: the bridge can omit fields it hasn't queried yet and the brain still gets a valid model.

nearby_blocks class-attribute instance-attribute 

nearby_blocks: list[NearbyBlock] = Field(default_factory=list)

Blocks within the configured scan radius, sorted by distance.

nearby_entities class-attribute instance-attribute 

nearby_entities: list[NearbyEntity] = Field(default_factory=list)

Mobs and players the bot can detect.

chat_log class-attribute instance-attribute 

chat_log: list[ChatMessage] = Field(default_factory=list)

Last N chat messages (configurable in MinecraftEnvironment).

render 

render() -> str

Compact text summary injected into the LLM prompt.

Source code in src/simulatecraft/minecraft/observations.py 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
def render(self) -> str:
    """Compact text summary injected into the LLM prompt."""
    lines: list[str] = [
        f"Tick {self.tick} | Pos {self.position} | Biome: {self.biome}",
        f"Health {self.stats.health}/20 | Food {self.stats.food}/20 "
        f"| Time {self.stats.time_of_day} | Rain: {self.stats.is_raining}",
    ]

    if self.equipped_item:
        lines.append(f"Holding: {self.equipped_item}")

    if self.inventory:
        inv_summary = ", ".join(f"{item.count}x {item.name}" for item in self.inventory[:12])
        if len(self.inventory) > 12:
            inv_summary += f" ... (+{len(self.inventory) - 12} more)"
        lines.append(f"Inventory: {inv_summary}")

    if self.nearby_blocks:
        block_summary = ", ".join(
            f"{b.name}@({b.x},{b.y},{b.z})" for b in self.nearby_blocks[:8]
        )
        lines.append(f"Nearby blocks: {block_summary}")

    if self.nearby_entities:
        ent_summary = ", ".join(
            f"{e.name}({e.entity_type}) ~{e.distance:.1f}m" for e in self.nearby_entities[:6]
        )
        lines.append(f"Nearby entities: {ent_summary}")

    if self.craftable:
        craft_summary = ", ".join(r.item_name for r in self.craftable[:6])
        lines.append(f"Craftable: {craft_summary}")

    if self.chat_log:
        lines.append("Recent chat:")
        for msg in self.chat_log[-4:]:
            prefix = f"<{msg.sender}> " if msg.sender else "[server] "
            lines.append(f"  {prefix}{msg.text}")

    if self.current_goal:
        lines.append(f"Current goal: {self.current_goal}")

    return "\n".join(lines)

NearbyBlock

Bases: BaseModel

A block within the scan radius.

NearbyEntity

Bases: BaseModel

A mob or player the bot can see.

Vec3

Bases: BaseModel

3-D float coordinate.