PlatformIO Driver

PlatformIO driver is the core abstraction of MaiBot's platform IO layer. This document details the driver interface definition, core types, and how to implement and register a custom driver.

PlatformIODriver Base Class

PlatformIODriver (defined in src/platform_io/drivers/base.py) is the abstract base class that all platform IO drivers must inherit:

class PlatformIODriver(ABC):
    def __init__(self, descriptor: DriverDescriptor) -> None: ...

    @property
    def descriptor(self) -> DriverDescriptor: ...

    @property
    def driver_id(self) -> str: ...

    def set_inbound_handler(self, handler: InboundHandler) -> None: ...
    def clear_inbound_handler(self) -> None: ...

    async def emit_inbound(self, envelope: InboundMessageEnvelope) -> bool: ...
    async def start(self) -> None: ...
    async def stop(self) -> None: ...

    @abstractmethod
    async def send_message(
        self, message: "SessionMessage", route_key: RouteKey, metadata: Optional[Dict[str, Any]] = None
    ) -> DeliveryReceipt: ...

Required Methods

Method	Description
send_message(message, route_key, metadata)	Send messages through specific driver, return DeliveryReceipt. This is the only abstract method that must be implemented

Optional Override Hooks

Method	Description
start()	Start driver lifecycle (default empty implementation)
stop()	Stop driver lifecycle (default empty implementation)

Inbound Message Reporting

When the driver receives external platform messages, it needs to construct InboundMessageEnvelope and call emit_inbound() to report:

envelope = InboundMessageEnvelope(
    route_key=RouteKey(platform="qq", account_id="123456"),
    driver_id=self.driver_id,
    driver_kind=self.descriptor.kind,
    external_message_id="platform_msg_id",
    session_message=parsed_message,
)
accepted = await self.emit_inbound(envelope)

emit_inbound returns True indicating the message is accepted by Broker and continues forwarding, False indicating rejection (inbound callback not configured or message filtered by deduplication).

Core Types

RouteKey — Routing Key

RouteKey is the unique key for routing decisions, using a three-layer structure:

@dataclass(frozen=True, slots=True)
class RouteKey:
    platform: str                           # Platform name, like "qq", "discord"
    account_id: Optional[str] = None        # Robot account ID
    scope: Optional[str] = None             # Additional routing scope

Routing resolution follows "from most specific to most general" fallback order: platform + account_id + scope → platform + account_id → platform. You can get the complete fallback chain through the resolution_order() method:

key = RouteKey(platform="qq", account_id="123", scope="group_456")
key.resolution_order()
# → [RouteKey("qq", "123", "group_456"), RouteKey("qq", "123", None), RouteKey("qq", None, None)]

The to_dedupe_scope() method generates a cross-driver shared deduplication scope string, formatted as platform:account_id:scope.

InboundMessageEnvelope — Inbound Message Encapsulation

@dataclass(slots=True)
class InboundMessageEnvelope:
    route_key: RouteKey                                # Inbound routing key
    driver_id: str                                     # Producing driver ID
    driver_kind: DriverKind                            # Driver type
    external_message_id: Optional[str] = None          # Platform-side message ID (for deduplication)
    dedupe_key: Optional[str] = None                   # Explicit deduplication key
    session_message: Optional["SessionMessage"] = None # Normalized message
    payload: Optional[Dict[str, Any]] = None           # Original dictionary payload
    metadata: Dict[str, Any] = field(default_factory=dict)

The priority of deduplication keys is: dedupe_key > external_message_id > session_message.message_id. Broker will not guess deduplication keys based on payload content to avoid misjudging different messages with the same content as duplicates.

DeliveryReceipt — Outbound Receipt

@dataclass(slots=True)
class DeliveryReceipt:
    internal_message_id: str                    # Internal message ID
    route_key: RouteKey                         # Delivery routing key
    status: DeliveryStatus                      # Delivery status
    driver_id: Optional[str] = None             # Processing driver ID
    driver_kind: Optional[DriverKind] = None    # Processing driver type
    external_message_id: Optional[str] = None   # Platform-side message ID
    error: Optional[str] = None                 # Error information
    metadata: Dict[str, Any] = field(default_factory=dict)

DeliveryStatus enumeration values:

Status	Description
PENDING	Pending sending
SENT	Sent
FAILED	Sending failed
DROPPED	Dropped

DriverDescriptor — Driver Description

@dataclass(frozen=True, slots=True)
class DriverDescriptor:
    driver_id: str                                # Globally unique driver ID
    kind: DriverKind                              # Driver type (LEGACY / PLUGIN)
    platform: str                                 # Platform name
    account_id: Optional[str] = None              # Account ID
    scope: Optional[str] = None                   # Routing scope
    plugin_id: Optional[str] = None               # Associated plugin ID
    metadata: Dict[str, Any] = field(default_factory=dict)

DriverKind enumeration distinguishes driver sources: LEGACY indicates built-in drivers, PLUGIN indicates drivers provided by plugins.

Implement and Register Driver

Complete Example

from src.platform_io.drivers.base import PlatformIODriver
from src.platform_io.types import (
    DeliveryReceipt, DeliveryStatus, DriverDescriptor, DriverKind,
    InboundMessageEnvelope, RouteKey, RouteBinding,
)

class MyDriver(PlatformIODriver):
    async def start(self) -> None:
        # Initialize connection
        self._connection = await connect_to_platform()

    async def stop(self) -> None:
        # Clean up connection
        await self._connection.close()

    async def send_message(self, message, route_key, metadata=None):
        try:
            platform_msg_id = await self._connection.send(
                target=route_key.account_id,
                content=message.plain_text,
            )
            return DeliveryReceipt(
                internal_message_id=message.message_id,
                route_key=route_key,
                status=DeliveryStatus.SENT,
                driver_id=self.driver_id,
                driver_kind=self.descriptor.kind,
                external_message_id=platform_msg_id,
            )
        except Exception as exc:
            return DeliveryReceipt(
                internal_message_id=message.message_id,
                route_key=route_key,
                status=DeliveryStatus.FAILED,
                driver_id=self.driver_id,
                driver_kind=self.descriptor.kind,
                error=str(exc),
            )

Registration and Route Binding

from src.platform_io.manager import get_platform_io_manager

manager = get_platform_io_manager()

# Descriptor
descriptor = DriverDescriptor(
    driver_id="my_driver.discord",
    kind=DriverKind.PLUGIN,
    platform="discord",
)

# Register
driver = MyDriver(descriptor)
await manager.add_driver(driver)

# Bind send route
manager.bind_send_route(RouteBinding(
    route_key=descriptor.route_key,
    driver_id=driver.driver_id,
    driver_kind=DriverKind.PLUGIN,
))

# Bind receive route
manager.bind_receive_route(RouteBinding(
    route_key=descriptor.route_key,
    driver_id=driver.driver_id,
    driver_kind=DriverKind.PLUGIN,
))

Use add_driver / remove_driver when Broker is running, use register_driver / unregister_driver when not running. Route binding supports priority sorting of multiple drivers under the same route key through the priority field.