plugin-development.md

Plugin Development Guide

Plugins are the primary extension mechanism in MindPy. They let you package custom behaviors, event handlers, and new capabilities into self-contained, reusable modules with automatic discovery and lifecycle management.

---

Quick start

from mindpy.plugins import Plugin, PluginMetadata
from mindpy.events import Event

class GreeterPlugin(Plugin):
    """Says hello to every player who joins."""

    @property
    def metadata(self) -> PluginMetadata:
        return PluginMetadata(
            name="greeter",
            version="1.0.0",
            description="Greet players when they join",
            author="You",
            dependencies=[],
        )

    async def on_enable(self) -> None:
        """Called when the plugin is loaded. Subscribe to events here."""
        self._token = self.event_bus.subscribe(
            "player.joined", self._on_player_join
        )
        self.logger.info("Greeter plugin enabled")

    async def on_disable(self) -> None:
        """Called when the plugin is unloaded. Clean up here."""
        self._token.cancel()
        self.logger.info("Greeter plugin disabled")

    async def _on_player_join(self, event: Event) -> None:
        name = event.data.get("username", "stranger")
        await self.bot.chat(f"Welcome, {name}! 👋")

---

Plugin base class

Every plugin extends mindpy.plugins.Plugin:

class Plugin(ABC):
    # Injected by PluginManager after load
    bot: Bot
    event_bus: EventBus
    config: Config
    logger: Logger

    @property
    @abstractmethod
    def metadata(self) -> PluginMetadata: ...

    @abstractmethod
    async def on_enable(self) -> None: ...

    @abstractmethod
    async def on_disable(self) -> None: ...

    # Optional hooks
    async def on_reload(self) -> None:
        await self.on_disable()
        await self.on_enable()

---

PluginMetadata

from mindpy.plugins import PluginMetadata

PluginMetadata(
    name="my_plugin",           # unique identifier (snake_case)
    version="1.0.0",            # semver
    description="Does X",
    author="You",               # optional
    dependencies=["other_plugin"],   # list of plugin names to load first
    min_mindpy_version="0.2.0", # optional version gate
)

---

Plugin lifecycle

PluginManager.load_plugin(plugin)
  │
  ├─ Inject: bot, event_bus, config, logger
  ├─ Resolve and load dependencies first
  └─ Call plugin.on_enable()
       │
       ▼
   [plugin is active]
       │
  PluginManager.reload_plugin("name")
  └─ Call plugin.on_reload()    (default: disable → enable)
       │
  PluginManager.unload_plugin("name")
  └─ Call plugin.on_disable()

Events published during the lifecycle:

Event	When
plugin.loaded	After on_enable() succeeds
plugin.unloaded	After on_disable() succeeds
plugin.error	If on_enable() or on_disable() raises

---

Using PluginManager

from mindpy.plugins import PluginManager

pm = bot.plugin_manager   # already available on Bot

# Load a plugin instance
await pm.load_plugin(GreeterPlugin())

# Load from a file path (auto-discovery)
await pm.load_from_file("plugins/greeter.py")

# Load all plugins in a directory
await pm.load_directory("plugins/")

# Get a running plugin
greeter = pm.get_plugin("greeter")

# Reload (calls on_disable → on_enable)
await pm.reload_plugin("greeter")

# Unload
await pm.unload_plugin("greeter")

# List all loaded plugins
for name, plugin in pm.list_plugins().items():
    print(f"{name} v{plugin.metadata.version}")

---

Auto-discovery

MindPy scans the configured plugins/ directory at startup. Place plugin files there and they will be loaded automatically:

plugins/
├── greeter.py          # single-file plugin
├── mining_assistant/   # package plugin
│   ├── __init__.py     # must export a Plugin subclass
│   └── utils.py
└── combat_helper.py

Single-file plugins must define exactly one Plugin subclass in the module. Package plugins must export a Plugin subclass from __init__.py.

---

Using the @handler decorator

For class-based event subscriptions, use the @handler decorator and register_handlers():

from mindpy.plugins import Plugin, PluginMetadata
from mindpy.events import handler, Event
from mindpy.events.event import EventPriority

class ChatLogPlugin(Plugin):
    @property
    def metadata(self) -> PluginMetadata:
        return PluginMetadata(name="chat_log", version="1.0.0",
                              description="Log all chat to file")

    async def on_enable(self) -> None:
        # Register all @handler-decorated methods on self
        self.event_bus.register_handlers(self)

    async def on_disable(self) -> None:
        # Unregister all methods registered via register_handlers
        self.event_bus.unregister_handlers(self)   # if implemented
        # Or keep SubscriptionTokens and cancel them manually

    @handler("chat.message", priority=EventPriority.LOW)
    async def on_chat(self, event: Event) -> None:
        with open("chat.log", "a", encoding="utf-8") as f:
            f.write(f"{event.timestamp.isoformat()}  {event.data.get('raw', '')}\n")

    @handler("player.joined")
    async def on_join(self, event: Event) -> None:
        self.logger.info(f"Player joined: {event.data.get('username')}")

---

Accessing bot state and sending messages

Plugins have full access to the bot through self.bot:

async def on_enable(self) -> None:
    self.event_bus.subscribe("chat.message", self._handle_command)

async def _handle_command(self, event: Event) -> None:
    raw = event.data.get("raw", "")

    if "!pos" in raw:
        x, y, z = self.bot.state.position
        await self.bot.chat(f"Position: ({x:.0f}, {y:.0f}, {z:.0f})")

    elif "!health" in raw:
        hp = self.bot.state.health
        food = self.bot.state.hunger
        await self.bot.chat(f"HP: {hp:.1f}/20  Food: {food}/20")

    elif "!goto" in raw:
        # Parse coordinates from message
        parts = raw.split()
        try:
            _, x, y, z = parts
            await self.bot.move_to(float(x), float(y), float(z))
            await self.bot.chat(f"Moving to ({x}, {y}, {z})")
        except (ValueError, IndexError):
            await self.bot.chat("Usage: !goto <x> <y> <z>")

---

Plugin configuration

Plugins can read from the shared Config via self.config:

# config.yaml
plugins:
  greeter:
    message: "Welcome, {name}! Have fun!"
    enabled: true
  chat_log:
    file: logs/chat.log

async def on_enable(self) -> None:
    self._message = self.config.get(
        "plugins.greeter.message",
        "Welcome, {name}!"
    )

---

Publishing custom events

Plugins can publish their own events for other plugins to consume:

from mindpy.events import Event
from mindpy.events.event import EventPriority

# Define your event type as a constant
PLAYER_GREETED = "greeter.player_greeted"

async def _on_player_join(self, event: Event) -> None:
    name = event.data.get("username", "stranger")
    await self.bot.chat(self._message.format(name=name))

    # Notify other plugins
    await self.event_bus.publish(Event(
        event_type=PLAYER_GREETED,
        data={"username": name},
        source="greeter",
    ))

---

Example: full-featured plugin

"""
AutoRespawn plugin — automatically respawns the bot when it dies.
"""
import asyncio
from mindpy.plugins import Plugin, PluginMetadata
from mindpy.events import Event


class AutoRespawnPlugin(Plugin):
    """Respawns the bot automatically after death, with a configurable delay."""

    @property
    def metadata(self) -> PluginMetadata:
        return PluginMetadata(
            name="auto_respawn",
            version="1.1.0",
            description="Automatically respawn after death",
            author="MindPy Team",
            dependencies=[],
        )

    async def on_enable(self) -> None:
        delay = self.config.get("plugins.auto_respawn.delay", 3.0)
        self._delay = float(delay)
        self._token = self.event_bus.subscribe("bot.died", self._on_death)
        self.logger.info(f"AutoRespawn enabled (delay={self._delay}s)")

    async def on_disable(self) -> None:
        self._token.cancel()

    async def _on_death(self, event: Event) -> None:
        self.logger.info(f"Bot died — respawning in {self._delay}s")
        await asyncio.sleep(self._delay)
        # Send Respawn action (Client Status 0x00)
        await self.event_bus.publish(Event(
            event_type="bot.respawn_requested",
            data={},
            source="auto_respawn",
        ))
        await self.bot.chat("I'm back! ♻️")

---

Testing plugins

import asyncio
import pytest
from unittest.mock import AsyncMock, MagicMock
from mindpy.events import EventBus, Event

from plugins.greeter import GreeterPlugin


@pytest.mark.asyncio
async def test_greeter_sends_welcome() -> None:
    bus = EventBus()
    await bus.start()

    mock_bot = MagicMock()
    mock_bot.chat = AsyncMock()
    mock_bot.state = MagicMock()

    plugin = GreeterPlugin()
    plugin.bot = mock_bot
    plugin.event_bus = bus
    plugin.config = MagicMock()
    plugin.config.get.return_value = "Welcome, {name}!"
    plugin.logger = MagicMock()

    await plugin.on_enable()

    # Simulate a player join event
    await bus.publish(Event(
        event_type="player.joined",
        data={"username": "Alice"},
    ))
    await asyncio.sleep(0.05)   # let bus worker drain

    mock_bot.chat.assert_called_once()
    call_arg = mock_bot.chat.call_args[0][0]
    assert "Alice" in call_arg

    await plugin.on_disable()
    await bus.stop()

---

Best practices

Rule	Why
Always cancel subscriptions in on_disable	Prevents ghost handlers after unload
Use SubscriptionToken or @handler	Makes cleanup explicit
Read config in on_enable, not __init__	Ensures config is injected before access
Publish events for side effects	Lets other plugins react without tight coupling
Keep handlers fast	Slow handlers block the event bus worker
Use asyncio.create_task for long work	Keeps the handler returning quickly
Respect min_mindpy_version	Prevents crashes on older framework versions
Write tests with a mock bot	Catches regressions before shipping