Mineflayer IPC bridge — Python side.
Architecture
A Node.js process runs bot/bot.js which connects to Minecraft via Mineflayer.
This module spawns that process and communicates with it over a local TCP socket
using newline-delimited JSON (one JSON object per line).
Python sends → {"id": "", "method": "", "params": {...}}
Node responds ← {"id": "", "result": {...}} or {"id": "", "error": "..."}
Node also pushes unsolicited events:
← {"event": "", "data": {...}}
Usage
bridge = MinecraftBridge(host="localhost", minecraft_port=25565,
username="SimBot", bot_script=None)
await bridge.connect()
state = await bridge.get_state()
await bridge.perform_action({"kind": "chat", "text": "hello!"})
await bridge.close()
bot_script defaults to the bundled bot/bot.js.
Bases: Exception
Raised when the bot process crashes or returns an error response.
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
| 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
|
async
Spawn the Node bot and wait until it signals it has joined the server.
Source code in src/simulatecraft/minecraft/connection.py
| 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)
|
async
Gracefully shut down the bot and the subprocess.
Source code in src/simulatecraft/minecraft/connection.py
| 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()
|
async
call(method: str, **params: Any) -> Any
Send an RPC request and await its response.
Source code in src/simulatecraft/minecraft/connection.py
| 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
|
async
get_state() -> dict[str, Any]
Return a full world-state snapshot from the bot.
Source code in src/simulatecraft/minecraft/connection.py
| 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(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
| 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(*, 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
| 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,
)
|