simulatecraft.minecraft.env

simulatecraft.minecraft.env

MinecraftEnvironment — bridges SimulateCraft's Environment interface to Mineflayer.

One MinecraftEnvironment manages one or more bots (one per registered agent). Each agent gets its own MinecraftBridge connection to its own bot process, so agents can be physically separate bots in the same Minecraft server.

Usage

env = MinecraftEnvironment(
    server_host="localhost",
    server_port=25565,
)
async with env:
    env.register_agent("alex", username="Alex")
    env.register_agent("bob",  username="Bob")
    runner = Runner(environment=env, config=RunnerConfig(tick_rate=1.0))
    runner.add_agent(Agent(id="alex", brain=LLMBrain(...)))
    runner.add_agent(Agent(id="bob",  brain=LLMBrain(...)))
    await runner.start()

AgentBotConfig

AgentBotConfig(username: str, password: str = '', ipc_port: int = _IPC_PORT_BASE, 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

def __init__(
    self,
    username: str,
    password: str = "",
    ipc_port: int = _IPC_PORT_BASE,
    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, map_pan_limit: int = 512)

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

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,
    map_pan_limit: int = 512,
) -> 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
    # Viewer pan/scan radius in blocks around home (half-side of square).
    self._map_pan_limit: int = max(128, min(int(map_pan_limit), 8192))
    self._world_settings: WorldSettings = load_settings()
    self._day_accum_ticks: float = 0.0
    self._last_day_advance: float | None = None
    self._map_preload_lock = asyncio.Lock()
    self._scout_ports: set[int] = set()

apply_world_rules

apply_world_rules() -> list[str]

Push gamerules / time to the live Minecraft server.

Source code in src/simulatecraft/minecraft/env.py

def apply_world_rules(self) -> list[str]:
    """Push gamerules / time to the live Minecraft server."""
    self.reload_world_settings()
    try:
        return apply_rules_via_rcon(self._world_settings.rules)
    except Exception as exc:
        log.warning("Applying world rules failed: %s", exc)
        return []

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 35670 (avoids Minecraft 25565 / RCON 25575).

Source code in src/simulatecraft/minecraft/env.py

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 35670
    (avoids Minecraft 25565 / RCON 25575).
    """
    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

async def connect(self) -> None:
    """Spawn all bots and wait for them to join the server."""
    self.apply_world_rules()
    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

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:
        # Soften join lag so existing agents don't fall through unloaded chunks.
        await self._protect_agents_for_join()
        await self._connect_one(agent_id)
        # Only refresh the new bot — fetching everyone spikes the server and
        # is a common trigger for mass deaths when many agents are online.
        if hasattr(self, "_obs_cache"):
            try:
                self._obs_cache[agent_id] = await self._fetch_state(agent_id)
            except Exception as exc:
                log.warning("State fetch for new agent '%s' failed: %s", agent_id, exc)
        else:
            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

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

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

async def prepare_tick(self) -> None:
    """Refresh bot observations before the runner asks each agent to decide."""
    self._advance_custom_daylight()
    self._enforce_boundaries()
    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

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

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

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, *, clamp: bool = True) -> 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

async def fetch_map(
    self, origin_x: int, origin_z: int, size: int | None = None, *, clamp: bool = True
) -> dict[str, Any]:
    """Scan a top-down map tile for the viewer (also used by WS pan requests)."""
    tile = max(16, min(int(size or self._map_size), 128))
    ox = int(origin_x)
    oz = int(origin_z)
    if clamp and 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))
    empty = {
        "origin_x": ox,
        "origin_z": oz,
        "width": tile,
        "height": tile,
        "pixels": "",
        "coverage": 0.0,
    }
    if not self._bridges:
        return empty
    bridge = next(iter(self._bridges.values()))
    try:
        result = await bridge.get_map(ox, oz, tile)
    except Exception as exc:
        log.warning("get_map failed at %s,%s: %s", ox, oz, exc)
        return empty
    if not isinstance(result, dict):
        return empty
    result.setdefault("origin_x", ox)
    result.setdefault("origin_z", oz)
    result.setdefault("width", tile)
    result.setdefault("height", tile)
    self._map_cache = result
    self._map_origin = (ox, oz)
    return result

map_tile_origins staticmethod

map_tile_origins(min_x: float, max_x: float, min_z: float, max_z: float, *, tile: int = 128) -> list[tuple[int, int]]

Grid of tile origins covering an XZ AABB (inclusive).

Source code in src/simulatecraft/minecraft/env.py

@staticmethod
def map_tile_origins(
    min_x: float,
    max_x: float,
    min_z: float,
    max_z: float,
    *,
    tile: int = 128,
) -> list[tuple[int, int]]:
    """Grid of tile origins covering an XZ AABB (inclusive)."""
    tile = max(16, int(tile))
    x0 = math.floor(min(min_x, max_x) / tile) * tile
    x1 = math.floor(max(min_x, max_x) / tile) * tile
    z0 = math.floor(min(min_z, max_z) / tile) * tile
    z1 = math.floor(max(min_z, max_z) / tile) * tile
    return [(ox, oz) for ox in range(x0, x1 + 1, tile) for oz in range(z0, z1 + 1, tile)]

preload_map_region async

preload_map_region(min_x: float, max_x: float, min_z: float, max_z: float, *, y: float | None = None, settle_seconds: float = 2.0, max_tiles: int = 512, workers: int = 8, use_agents: bool = False, on_tile: Callable[[dict[str, Any], int, int], Awaitable[None] | None] | None = None) -> dict[str, Any]

Scan an XZ box with temporary scout bots (parallel), stream via on_tile.

By default only scout clients are teleported — your simulation agents stay put. Pass use_agents=True to also draft connected agents as workers. Caps at max_tiles (default 512 ≈ 2896×2896 blocks at tile 128).

Source code in src/simulatecraft/minecraft/env.py

async def preload_map_region(
    self,
    min_x: float,
    max_x: float,
    min_z: float,
    max_z: float,
    *,
    y: float | None = None,
    settle_seconds: float = 2.0,
    max_tiles: int = 512,
    workers: int = 8,
    use_agents: bool = False,
    on_tile: Callable[[dict[str, Any], int, int], Awaitable[None] | None] | None = None,
) -> dict[str, Any]:
    """Scan an XZ box with temporary scout bots (parallel), stream via ``on_tile``.

    By default only scout clients are teleported — your simulation agents stay
    put. Pass ``use_agents=True`` to also draft connected agents as workers.
    Caps at ``max_tiles`` (default 512 ≈ 2896×2896 blocks at tile 128).
    """
    if not self._bridges:
        raise RuntimeError("No bots connected — spawn an agent before preloading the map")
    if self._map_preload_lock.locked():
        raise RuntimeError("Map preload already in progress")

    async with self._map_preload_lock:
        return await self._preload_map_region_locked(
            min_x,
            max_x,
            min_z,
            max_z,
            y=y,
            settle_seconds=settle_seconds,
            max_tiles=max_tiles,
            workers=workers,
            use_agents=use_agents,
            on_tile=on_tile,
        )

snapshot

snapshot() -> Snapshot

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

Source code in src/simulatecraft/minecraft/env.py

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,
            "boundaries": self._world_settings.boundaries.model_dump(mode="json"),
            "rules": self._world_settings.rules.model_dump(mode="json"),
            "world_info": self._world_settings.world.model_dump(mode="json"),
        },
    )