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 = 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

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

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

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

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

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

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."""
    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) -> 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
) -> 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

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,
        },
    )