Complete API documentation for MindPy v0.2.0.
from mindpy import (
Bot, # main bot class
BotState, # in-game state snapshot
EventBus, # event publish/subscribe
Event, # event data model
EventTypes, # standard event type constants
EventPriority, # handler priority enum
SubscriptionToken, # returned by subscribe()
handler, # @handler class-level decorator
Config, # configuration manager
Plugin, # plugin base class
PluginManager, # plugin lifecycle manager
)The main entry point. Manages connection, event bus, and bot control.
class Bot:
def __init__(
self,
host: str = "localhost",
port: int = 25565,
username: str = "MindPyBot",
password: str | None = None, # unused, for compat
config: Config | None = None,
auth_profile: AuthProfile | None = None, # for online mode
protocol_version: int = 765, # 1.20.4
online_mode: bool = False,
view_distance: int = 10,
) -> None: ...| Attribute | Type | Description |
|---|---|---|
host |
str |
Server host |
port |
int |
Server port |
username |
str |
Bot username |
state |
BotState |
Current in-game state |
event_bus |
EventBus |
The event bus |
plugin_manager |
PluginManager |
Plugin manager |
config |
Config |
Active configuration |
protocol_version |
int |
Minecraft protocol version |
logger |
Logger |
loguru logger |
Open TCP connection, perform the full login sequence (Handshaking → Login → Play), start the event bus worker, and publish BOT_CONNECTED.
Raises ConnectionError on failure.
Close the connection, publish BOT_DISCONNECTED, then stop the event bus. Safe to call even if not connected.
Exponential-backoff retry loop (up to max_reconnect_attempts). Publishes BOT_RECONNECTING before each attempt.
Block until disconnected. Calls connect() first if not already connected. Automatically triggers reconnect() if the connection drops.
Send a chat message. Messages longer than 256 characters are silently truncated. No-op if not connected.
Alias for chat().
Send a PlayerPositionAndLook serverbound packet. Updates bot.state optimistically.
True if the bot is in the PLAY state.
Returns (x, y, z) from bot.state.
Returns bot.state.health.
Returns bot.state.hunger.
Fluent decorator to subscribe a coroutine to an event:
@bot.on("chat.message", priority=EventPriority.HIGH)
async def on_chat(event: Event) -> None:
print(event.data["raw"])Wildcards are supported: "bot.*", "player.*".
async with Bot("localhost") as bot:
await bot.run()
# connect() called on enter, disconnect() called on exit@dataclass
class BotState:
connected: bool = False
health: float = 20.0
hunger: int = 20
saturation: float = 5.0
experience: int = 0
level: int = 0
x: float = 0.0
y: float = 64.0
z: float = 0.0
yaw: float = 0.0
pitch: float = 0.0
on_ground: bool = True
dimension: str = "minecraft:overworld"
game_mode: int = 0
entity_id: int = -1
is_hardcore: bool = False
@property
def position(self) -> tuple[float, float, float]: ...from mindpy.events import EventBus, Event
bus = EventBus()
await bus.start() # starts worker task
await bus.stop() # drains queue then stopsSubscribe handler to events matching event_type. Wildcards supported (fnmatch semantics).
Subscribe to all events (equivalent to subscribe("*", ...)).
Remove a handler.
Enqueue an event. If the bus is not running, dispatches inline.
Thread-safe publish from synchronous code (schedules a task on the running loop).
Block until a matching event arrives. Raises asyncio.TimeoutError if timeout expires.
Inspect obj for methods decorated with @handler and subscribe them.
Count registered handlers (optionally filtered by type).
Remove all handlers.
token = bus.subscribe("chat.message", handler)
token.cancel() # unsubscribe
with token: # context manager
...
# automatically cancelled on exit@dataclass
class Event:
event_type: str
data: dict[str, Any] = field(default_factory=dict)
priority: EventPriority = EventPriority.NORMAL
timestamp: datetime = ... # auto: datetime.now(UTC)
event_id: str = ... # auto: uuid4()
source: str = "unknown"
def get(self, key: str, default=None) -> Any: ...
def has(self, key: str) -> bool: ...
def to_dict(self) -> dict: ...
@classmethod
def from_dict(cls, data: dict) -> "Event": ...Standard event type string constants:
from mindpy.events.event import EventTypes
# Bot lifecycle
EventTypes.BOT_CONNECTED # "bot.connected"
EventTypes.BOT_DISCONNECTED # "bot.disconnected"
EventTypes.BOT_RECONNECTING # "bot.reconnecting"
EventTypes.BOT_ERROR # "bot.error"
EventTypes.BOT_SPAWNED # "bot.spawned"
EventTypes.BOT_DIED # "bot.died"
EventTypes.BOT_POSITION_CHANGED # "bot.position_changed"
EventTypes.BOT_HEALTH_CHANGED # "bot.health_changed"
# Chat
EventTypes.CHAT_MESSAGE # "chat.message"
EventTypes.PLAYER_CHAT # "player.chat" (legacy alias)
# World
EventTypes.CHUNK_LOADED # "chunk.loaded"
EventTypes.CHUNK_UNLOADED # "chunk.unloaded"
EventTypes.WORLD_CHANGED # "world.changed"
# Entities
EventTypes.ENTITY_SPAWNED # "entity.spawned"
EventTypes.ENTITY_DESPAWNED # "entity.despawned"
EventTypes.ENTITY_MOVED # "entity.moved"
EventTypes.ENTITY_DAMAGED # "entity.damaged"
EventTypes.ENTITY_DIED # "entity.died"
# Players
EventTypes.PLAYER_JOINED # "player.joined"
EventTypes.PLAYER_LEFT # "player.left"
EventTypes.PLAYER_DEATH # "player.death"
EventTypes.PLAYER_RESPAWN # "player.respawn"
# Tasks & Goals
EventTypes.TASK_STARTED # "task.started"
EventTypes.TASK_FINISHED # "task.finished"
EventTypes.TASK_FAILED # "task.failed"
EventTypes.TASK_CANCELLED # "task.cancelled"
EventTypes.GOAL_CREATED # "goal.created"
EventTypes.GOAL_COMPLETED # "goal.completed"
EventTypes.GOAL_FAILED # "goal.failed"
# AI
EventTypes.AI_DECISION # "ai.decision"
EventTypes.AI_REFLECTION # "ai.reflection"
EventTypes.AI_TOOL_CALL # "ai.tool_call"
# Plugins
EventTypes.PLUGIN_LOADED # "plugin.loaded"
EventTypes.PLUGIN_UNLOADED # "plugin.unloaded"
EventTypes.PLUGIN_ERROR # "plugin.error"from mindpy.events.event import EventPriority
EventPriority.CRITICAL # 0 — dispatched first
EventPriority.HIGH # 1
EventPriority.NORMAL # 2 — default
EventPriority.LOW # 3 — dispatched lastfrom mindpy.protocol.codec import ProtocolRegistry
codec = ProtocolRegistry.get(769) # exact match
codec = ProtocolRegistry.get(770) # fallback to 769
ProtocolRegistry.supported_versions() # → [47, 340, 754, 762, 765, 769]
ProtocolRegistry.is_supported(770) # → True (via fallback)
# Look up a packet class
pkt_cls = codec.registry.get(
ConnectionState.PLAY,
PacketDirection.CLIENTBOUND,
0x26, # KeepAlive in v769
)from mindpy.protocol.connection import MinecraftConnection
from mindpy.protocol.codec import ProtocolRegistry
import mindpy.protocol.versions # register all versions
codec = ProtocolRegistry.get(765)
conn = MinecraftConnection("mc.example.com", 25565, 765, codec.registry)
await conn.connect()
conn.enable_encryption(shared_secret_bytes)
conn.set_compression(threshold=256)
conn.set_state(ConnectionState.PLAY)
conn.start_reading() # start background read loop
await conn.send_packet(some_packet)
await conn.close()Register per-packet handlers:
from mindpy.protocol.versions.v769 import KeepAliveClientboundPacket
@conn.on_packet(KeepAliveClientboundPacket)
async def handle_ka(packet: KeepAliveClientboundPacket) -> None:
print(f"KeepAlive ID: {packet.keep_alive_id}")
# or programmatically
conn.add_handler(KeepAliveClientboundPacket, handle_ka)from mindpy.protocol.login import LoginOrchestrator, LoginError, ServerInfo
orchestrator = LoginOrchestrator(
connection=conn,
username="Bot",
auth_profile=profile, # None for offline mode
online_mode=True,
)
# Optional: ping first to check version
info: ServerInfo = await orchestrator.ping()
print(info.protocol_version, info.version_name, info.latency_ms)
# Login: transitions conn to PLAY state and starts read loop
await orchestrator.login()from mindpy.protocol.auth import MicrosoftAuth, AuthProfile
async with MicrosoftAuth() as auth:
profile: AuthProfile = await auth.device_flow_auth()
# Prints a URL + code; sign in within 15 minutes
# profile fields:
profile.username # str
profile.uuid # str (with hyphens)
profile.access_token # str (Minecraft access token)from mindpy.protocol.types import PacketBuffer, VarInt
# Writing
buf = PacketBuffer()
buf.write_varint(123)
buf.write_string("hello")
buf.write_double(3.14)
buf.write_bool(True)
buf.write_long(99_999_999)
raw = buf.to_bytes()
# Reading
buf2 = PacketBuffer(raw)
val = buf2.read_varint()
s = buf2.read_string()
d = buf2.read_double()
b = buf2.read_bool()
l = buf2.read_long()
# VarInt standalone
encoded = VarInt.encode(300) # → b'\xac\x02'
decoded = PacketBuffer(encoded).read_varint() # → 300from mindpy.config import Config
cfg = Config("config.yaml") # load from file
cfg = Config() # defaults only
val = cfg.get("bot.host", "localhost") # dot-path access
cfg.set("bot.username", "Agent1")
cfg.save()
d = cfg.to_dict()from mindpy.tasks import TaskManager, BaseTask, TaskData, TaskStatus, TaskPriority
class MyTask(BaseTask):
@property
def name(self) -> str: return "my_task"
@property
def description(self) -> str: return "Does something"
async def execute(self, context) -> str:
# ...
return TaskStatus.COMPLETED.value
manager = TaskManager(event_bus)
await manager.start(num_workers=4)
task_id = await manager.submit_task(MyTask())
await manager.suspend_task(task_id)
await manager.resume_task(task_id)
await manager.cancel_task(task_id)
stats = manager.get_statistics()
await manager.stop()from mindpy.goals import GoalManager, Goal, GoalStatus, GoalPriority
gm = GoalManager(event_bus)
goal = gm.create_goal(
name="Mine Diamonds",
description="Mine 10 diamonds",
priority=GoalPriority.HIGH,
)
sub_goals = await gm.decompose_goal(goal.goal_id)
gm.update_goal_status(goal.goal_id, GoalStatus.IN_PROGRESS)
tree = gm.get_goal_tree(goal.goal_id)
pending = gm.get_goals_by_status(GoalStatus.PENDING)from mindpy.memory import MemoryManager
mm = MemoryManager()
# Working memory
mm.working_memory.add("key", "value")
mm.working_memory.get("key")
# Long-term memory
mm.long_term_memory.store_fact("home", (100, 64, 200))
mm.long_term_memory.get_fact("home")
# Conversation memory
mm.conversation_memory.add_message(role="user", content="Go mining")
mm.conversation_memory.get_recent(n=10)
await mm.save_all()
await mm.load_all()
stats = mm.get_statistics()from mindpy.llm import LLMManager, LLMMessage
llm = LLMManager()
llm.setup_openai(api_key="sk-...", model="gpt-4o")
llm.setup_anthropic(api_key="sk-ant-...", model="claude-3-5-sonnet-20241022")
llm.setup_gemini(api_key="AIza...", model="gemini-1.5-pro")
llm.setup_ollama(base_url="http://localhost:11434", model="llama3")
response = await llm.generate([
LLMMessage(role="system", content="You are a Minecraft bot."),
LLMMessage(role="user", content="What should I do?"),
])
print(response.content)from mindpy.ai import AIAgent, AgentContext
agent = AIAgent(
llm_manager=llm,
tool_registry=None, # optional
system_prompt="You are a Minecraft helper."
)
ctx = AgentContext(
position=(100, 64, 200),
health=18.0,
inventory=[],
nearby_entities=[],
chat_history=[],
)
# Text decision
text = await agent.decide(ctx, user_message="What now?")
# Tool-calling decision
tool_calls = await agent.decide_with_tools(ctx, user_message="Mine some wood")from mindpy.plugins import Plugin, PluginMetadata, PluginManager
class MyPlugin(Plugin):
@property
def metadata(self) -> PluginMetadata:
return PluginMetadata(
name="my_plugin",
version="1.0.0",
description="My custom plugin",
dependencies=[],
)
async def on_enable(self) -> None:
self.event_bus.subscribe("chat.message", self._on_chat)
async def on_disable(self) -> None:
pass
async def _on_chat(self, event: Event) -> None:
print(event.data["raw"])
pm = PluginManager(event_bus)
await pm.load_plugin(MyPlugin())
await pm.unload_plugin("my_plugin")from mindpy.world.chunks import ChunkManager
cm = ChunkManager()
state_id = cm.get_block(x=100, y=64, z=200) # → int (global palette ID)
cm.set_block(x=100, y=64, z=200, state_id=1) # 1 = stone
cm.unload_chunk(chunk_x=6, chunk_z=12)from mindpy.world.blocks import BlockManager
bm = BlockManager(chunk_manager)
bm.get_block_at(100, 64, 200)
bm.set_block_at(100, 64, 200, state_id=1)from mindpy.world.entities import EntityManager
em = EntityManager()
em.spawn_entity(entity_id=42, entity_type="zombie", x=100.0, y=64.0, z=200.0)
em.move_entity(entity_id=42, x=101.0, y=64.0, z=200.0)
nearby = em.entities_in_radius(x=100.0, y=64.0, z=200.0, radius=32.0)
em.despawn_entity(entity_id=42)from mindpy.world.spatial import SpatialIndex
idx = SpatialIndex(cell_size=16)
idx.insert(entity_id=1, x=100.0, y=64.0, z=200.0)
idx.update(entity_id=1, x=101.0, y=64.0, z=200.0)
idx.remove(entity_id=1)
nearby = idx.entities_in_radius(x=100.0, y=64.0, z=200.0, radius=32.0)from mindpy.navigation import PathFinder
pf = PathFinder()
path = await pf.find_path(
start=(100, 64, 200),
goal=(150, 64, 250),
is_walkable=lambda pos: True, # optional override
)
smooth = pf.smooth_path(path)from mindpy.navigation import WaypointManager, WaypointType
wm = WaypointManager()
wm.add_waypoint("home", (100, 64, 200), WaypointType.LOCATION)
wp = wm.get_waypoint("home")
nearest = wm.find_nearest_waypoint((110, 64, 210), max_distance=50.0)
wm.remove_waypoint("home")| Enum | Values |
|---|---|
EventPriority |
CRITICAL(0) HIGH(1) NORMAL(2) LOW(3) |
ConnectionState |
HANDSHAKING STATUS LOGIN CONFIGURATION PLAY |
PacketDirection |
CLIENTBOUND SERVERBOUND |
GoalStatus |
PENDING IN_PROGRESS COMPLETED FAILED ABORTED BLOCKED |
GoalPriority |
CRITICAL HIGH NORMAL LOW |
TaskStatus |
PENDING RUNNING SUSPENDED COMPLETED FAILED CANCELLED |
TaskPriority |
CRITICAL HIGH NORMAL LOW |
WaypointType |
LOCATION BASE RESOURCE DANGER CUSTOM |