Skip to content

API Reference

Auto-generated from source via mkdocstrings. Private names (_*) are filtered out. Prefer the public re-exports on mcbe_ws_sdk when writing host code.

Public package

mcbe_ws_sdk

Generic WebSocket gateway SDK for Minecraft Bedrock Edition.

Public surface

The gateway SDK exposes a dual-layer interface:

  • Low-level: subscribe to an :class:~mcbe_ws_sdk.gateway.events.EventBus keyed by :class:~mcbe_ws_sdk.gateway.events.WsEventType.
  • High-level: implement :class:~mcbe_ws_sdk.gateway.hook.ConnectionHook and :class:~mcbe_ws_sdk.gateway.sink.ResponseSink, then drive the stack through :class:~mcbe_ws_sdk.gateway.server_facade.McbeServerFacade.

The full connection lifetime, packet request abstraction and byte-safe command chunking are provided; the agent's LLM / message-broker concerns are the host's.

Modules:

  • addon –

    Addon bridge capability for the MCBE WebSocket SDK.

  • command –

    Command parsing package: registry + parsed-command value object.

  • config –
  • delivery –

    Outbound delivery adapter for MCBE WebSocket gateway.

  • errors –

    Public exception hierarchy for mcbe-ws-sdk.

  • flow –

    Flow control middleware for outbound MCBE command chunking.

  • gateway –

    MCBE WebSocket gateway SDK.

  • logging –

    Optional console logging helpers for hosts and examples.

  • profiles –

    Protocol profiles for MCBE wire compatibility layers.

  • protocol –

    Minecraft protocol message models.

Classes:

Functions:

  • enqueue_response –

    Put message onto state.response_queue, dropping the oldest if full.

  • configure_logging –

    Configure structlog + stdlib logging for compact console output.

AddonBridgeClient

Bases: Protocol

Face an Agent uses to reach the addon bridge.

Methods:

  • request –

    Issue a capability request and return the reassembled payload.

request async 

request(capability: str, payload: dict[str, Any]) -> dict[str, Any]

Issue a capability request and return the reassembled payload.

Source code in src/mcbe_ws_sdk/addon/service.py 
54
55
async def request(self, capability: str, payload: dict[str, Any]) -> dict[str, Any]:
    """Issue a capability request and return the reassembled payload."""

AddonBridgeService 

AddonBridgeService(settings: AddonBridgeSettings)

Manage the request/response lifecycle between Python and the addon.

Methods:

Source code in src/mcbe_ws_sdk/addon/service.py 
61
62
63
64
65
66
def __init__(self, settings: AddonBridgeSettings) -> None:
    self._settings = settings
    self._sessions: dict[UUID, AddonBridgeSession] = {}
    self._timeout_seconds = settings.timeout_seconds
    self._profile = settings.profile
    self._ui_chat_callback: UiChatCallback | None = None

create_client 

create_client(connection_id: UUID, send_command: CommandSender) -> AddonBridgeClient

Build a client bound to one connection.

Source code in src/mcbe_ws_sdk/addon/service.py 
68
69
70
71
72
73
74
def create_client(
    self,
    connection_id: UUID,
    send_command: CommandSender,
) -> AddonBridgeClient:
    """Build a client bound to one connection."""
    return ConnectionAddonBridgeClient(self, connection_id, send_command)

request_capability async 

request_capability(connection_id: UUID, capability: str, payload: dict[str, Any], send_command: CommandSender) -> dict[str, Any]

Send a bridge request and wait for the addon's chat callback.

Source code in src/mcbe_ws_sdk/addon/service.py 
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
async def request_capability(
    self,
    connection_id: UUID,
    capability: str,
    payload: dict[str, Any],
    send_command: CommandSender,
) -> dict[str, Any]:
    """Send a bridge request and wait for the addon's chat callback."""
    session = self._session_for(connection_id)
    request = session.create_request(capability=capability, payload=payload)
    command = encode_bridge_request(
        request_id=request.request_id,
        capability=capability,
        payload=payload,
        profile=self._profile,
    )
    payload_bytes = len(
        json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
    )
    # INFO stays free of payload/command content (may contain player data).
    logger.info(
        "bridge_request_outbound",
        request_id=request.request_id,
        capability=capability,
        payload_bytes=payload_bytes,
    )
    logger.debug(
        "bridge_request_outbound",
        connection_id=str(connection_id),
        request_id=request.request_id,
        capability=capability,
        payload=payload,
        command=command,
        timeout_seconds=self._timeout_seconds,
    )
    try:
        await send_command(command)
        try:
            result = await asyncio.wait_for(request.future, self._timeout_seconds)
        except TimeoutError as exc:
            logger.warning(
                "bridge_request_timeout",
                connection_id=str(connection_id),
                request_id=request.request_id,
                capability=capability,
                timeout_seconds=self._timeout_seconds,
            )
            raise BridgeTimeoutError(request.request_id) from exc
        logger.info(
            "bridge_request_resolved",
            connection_id=str(connection_id),
            request_id=request.request_id,
            capability=capability,
            result=result,
        )
        return result
    finally:
        session.cancel_request(request.request_id)

is_bridge_chat_message 

is_bridge_chat_message(sender: str, message: str) -> bool

True if the player chat message is an addon response fragment.

Source code in src/mcbe_ws_sdk/addon/service.py 
135
136
137
138
139
def is_bridge_chat_message(self, sender: str, message: str) -> bool:
    """True if the player chat message is an addon response fragment."""
    return sender == self._profile.bridge_sender and message.startswith(
        self._profile.bridge_response_prefix
    )

is_ui_chat_message 

is_ui_chat_message(sender: str, message: str) -> bool

True if the player chat message is a UI chat fragment.

Source code in src/mcbe_ws_sdk/addon/service.py 
141
142
143
144
145
def is_ui_chat_message(self, sender: str, message: str) -> bool:
    """True if the player chat message is a UI chat fragment."""
    return sender == self._profile.bridge_sender and message.startswith(
        self._profile.ui_chat_prefix
    )

handle_player_message async 

handle_player_message(connection_id: UUID, sender: str, message: str) -> AddonMessageResult

Route a routed message from the simulated player.

Source code in src/mcbe_ws_sdk/addon/service.py 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
async def handle_player_message(
    self, connection_id: UUID, sender: str, message: str
) -> AddonMessageResult:
    """Route a routed message from the simulated player."""
    if self.is_bridge_chat_message(sender, message):
        # Log every RESP chunk at INFO: these never reach the host hook because
        # the facade short-circuits bridge frames before on_player_message.
        logger.info(
            "bridge_chat_chunk_raw",
            connection_id=str(connection_id),
            sender=sender,
            message=message,
        )
        session = self._sessions.get(connection_id)
        if session is None:
            logger.warning(
                "bridge_chat_no_session",
                connection_id=str(connection_id),
            )
            return AddonMessageResult(handled=True)

        bridge_chunk = session.handle_chat_chunk(message)
        return AddonMessageResult(handled=True, bridge_chunk=bridge_chunk)

    if self.is_ui_chat_message(sender, message):
        logger.info(
            "ui_chat_chunk_raw",
            connection_id=str(connection_id),
            sender=sender,
            message=message,
        )
        session = self._session_for(connection_id)

        ui_chunk, ui_message = session.handle_ui_chat_chunk(message)
        if ui_message is not None and self._ui_chat_callback is not None:
            logger.info(
                "ui_chat_reassembled",
                connection_id=str(connection_id),
                player=ui_message.player_name,
                message_length=len(ui_message.message),
                callback_registered=True,
            )
            await self._ui_chat_callback(
                connection_id,
                ui_message.player_name,
                ui_message.message,
            )
        return AddonMessageResult(
            handled=True,
            ui_chunk=ui_chunk,
            ui_message=ui_message,
        )

    return AddonMessageResult(handled=False)

set_ui_chat_callback 

set_ui_chat_callback(callback: UiChatCallback) -> None

Register the UI chat message callback.

Source code in src/mcbe_ws_sdk/addon/service.py 
202
203
204
def set_ui_chat_callback(self, callback: UiChatCallback) -> None:
    """Register the UI chat message callback."""
    self._ui_chat_callback = callback

close_connection 

close_connection(connection_id: UUID) -> None

Tear down the per-connection session on disconnect.

Source code in src/mcbe_ws_sdk/addon/service.py 
206
207
208
209
210
def close_connection(self, connection_id: UUID) -> None:
    """Tear down the per-connection session on disconnect."""
    session = self._sessions.pop(connection_id, None)
    if session is not None:
        session.close()

ConnectionAddonBridgeClient 

ConnectionAddonBridgeClient(service: AddonBridgeService, connection_id: UUID, send_command: CommandSender)

Per-connection client bound to one bridge service instance.

Source code in src/mcbe_ws_sdk/addon/service.py 
223
224
225
226
227
228
229
230
231
def __init__(
    self,
    service: AddonBridgeService,
    connection_id: UUID,
    send_command: CommandSender,
) -> None:
    self._service = service
    self._connection_id = connection_id
    self._send_command = send_command

CommandRegistry 

CommandRegistry(commands_config: Mapping[str, str | Mapping[str, Any]] | None = None)

Registry of command prefixes, types, and aliases.

Methods:

  • resolve –

    Resolve a message to (command_type, content).

  • resolve_parsed –

    Resolve a message to a typed command, including which token matched.

  • add_alias –

    Register a dynamic alias for an existing command prefix.

  • remove_alias –

    Remove a previously registered alias.

  • get_command_config –

    Return the loaded config for a primary command prefix.

  • get_aliases –

    Return all aliases for a command prefix.

  • list_all_commands –

    List all commands as (prefix, type, aliases) tuples.

  • get_command_prefix –

    Return the primary prefix for a command type, if registered.

Source code in src/mcbe_ws_sdk/command/registry.py 
43
44
45
46
47
48
49
50
def __init__(
    self,
    commands_config: Mapping[str, str | Mapping[str, Any]] | None = None,
) -> None:
    self._commands: dict[str, MinecraftCommandConfig] = {}
    self._alias_map: dict[str, str] = {}  # alias -> primary prefix
    self._type_to_prefix: dict[str, str] = {}  # command type -> primary prefix
    self._load_commands(commands_config or {})

resolve 

resolve(message: str) -> tuple[str | None, str]

Resolve a message to (command_type, content).

Source code in src/mcbe_ws_sdk/command/registry.py 
88
89
90
91
92
93
def resolve(self, message: str) -> tuple[str | None, str]:
    """Resolve a message to ``(command_type, content)``."""
    parsed = self.resolve_parsed(message)
    if parsed is None:
        return None, message
    return parsed.type, parsed.content

resolve_parsed 

resolve_parsed(message: str) -> ParsedCommand | None

Resolve a message to a typed command, including which token matched.

Source code in src/mcbe_ws_sdk/command/registry.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
def resolve_parsed(self, message: str) -> ParsedCommand | None:
    """Resolve a message to a typed command, including which token matched."""
    for prefix, cmd_config in self._commands.items():
        if self._matches_token(message, prefix):
            content = message[len(prefix) :].strip()
            return ParsedCommand(
                type=cmd_config.type,
                content=content,
                prefix=prefix,
                raw=message,
            )

    for alias, main_prefix in self._alias_map.items():
        if self._matches_token(message, alias):
            content = message[len(alias) :].strip()
            cmd_config = self._commands[main_prefix]
            return ParsedCommand(
                type=cmd_config.type,
                content=content,
                prefix=main_prefix,
                raw=message,
                matched_alias=alias,
            )

    return None

add_alias 

add_alias(command_prefix: str, alias: str) -> bool

Register a dynamic alias for an existing command prefix.

Source code in src/mcbe_ws_sdk/command/registry.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
def add_alias(self, command_prefix: str, alias: str) -> bool:
    """Register a dynamic alias for an existing command prefix."""
    if command_prefix not in self._commands:
        logger.warning("add_alias_command_not_found", prefix=command_prefix)
        return False

    if alias in self._alias_map:
        logger.warning("add_alias_already_exists", alias=alias)
        return False

    self._alias_map[alias] = command_prefix
    old = self._commands[command_prefix]
    self._commands[command_prefix] = replace(old, aliases=old.aliases + (alias,))

    logger.info("alias_added", prefix=command_prefix, alias=alias)
    return True

remove_alias 

remove_alias(alias: str) -> bool

Remove a previously registered alias.

Source code in src/mcbe_ws_sdk/command/registry.py 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
def remove_alias(self, alias: str) -> bool:
    """Remove a previously registered alias."""
    if alias not in self._alias_map:
        logger.warning("remove_alias_not_found", alias=alias)
        return False

    main_prefix = self._alias_map.pop(alias)
    old = self._commands[main_prefix]
    self._commands[main_prefix] = replace(
        old, aliases=tuple(a for a in old.aliases if a != alias)
    )

    logger.info("alias_removed", prefix=main_prefix, alias=alias)
    return True

get_command_config 

get_command_config(prefix: str) -> MinecraftCommandConfig | None

Return the loaded config for a primary command prefix.

Source code in src/mcbe_ws_sdk/command/registry.py 
161
162
163
def get_command_config(self, prefix: str) -> MinecraftCommandConfig | None:
    """Return the loaded config for a primary command prefix."""
    return self._commands.get(prefix)

get_aliases 

get_aliases(command_prefix: str) -> tuple[str, ...]

Return all aliases for a command prefix.

Source code in src/mcbe_ws_sdk/command/registry.py 
165
166
167
168
def get_aliases(self, command_prefix: str) -> tuple[str, ...]:
    """Return all aliases for a command prefix."""
    cmd = self._commands.get(command_prefix)
    return cmd.aliases if cmd else ()

list_all_commands 

list_all_commands() -> list[tuple[str, str, tuple[str, ...]]]

List all commands as (prefix, type, aliases) tuples.

Source code in src/mcbe_ws_sdk/command/registry.py 
170
171
172
def list_all_commands(self) -> list[tuple[str, str, tuple[str, ...]]]:
    """List all commands as ``(prefix, type, aliases)`` tuples."""
    return [(prefix, config.type, config.aliases) for prefix, config in self._commands.items()]

get_command_prefix 

get_command_prefix(cmd_type: str) -> str | None

Return the primary prefix for a command type, if registered.

Source code in src/mcbe_ws_sdk/command/registry.py 
174
175
176
def get_command_prefix(self, cmd_type: str) -> str | None:
    """Return the primary prefix for a command type, if registered."""
    return self._type_to_prefix.get(cmd_type)

McbeOutboundDelivery 

McbeOutboundDelivery(*, connection_id: Any, send_payload: PayloadSender, settings: FlowControlSettings, log_raw_payloads: bool = False)

Chunk, throttle, send, and optionally log MCBE commandRequest payloads.

Methods:

  • send_payload –

    Send an already-built payload (subscribe, handshake, non-chunked paths).

  • send_tellraw –

    Send tellraw text and return the number of payloads actually transmitted.

  • send_scriptevent –

    Send scriptevent text and return the number of payloads actually transmitted.

  • send_raw_command –

    Send a raw command that must not be semantically split.

  • send_outbound_text –

    Route an OutboundText message through the delivery adapter.

  • send_system_notification –

    Route a SystemNotification through the delivery adapter.

  • send_chunked –

    Send payloads with inter-chunk delays.

Source code in src/mcbe_ws_sdk/delivery/outbound.py 
22
23
24
25
26
27
28
29
30
31
32
33
def __init__(
    self,
    *,
    connection_id: Any,
    send_payload: PayloadSender,
    settings: FlowControlSettings,
    log_raw_payloads: bool = False,
) -> None:
    self.connection_id = connection_id
    self._send_payload = send_payload
    self._flow = FlowControlMiddleware(settings)
    self._log_raw_payloads = log_raw_payloads

send_payload async 

send_payload(payload: str, source: str) -> None

Send an already-built payload (subscribe, handshake, non-chunked paths).

Source code in src/mcbe_ws_sdk/delivery/outbound.py 
39
40
41
async def send_payload(self, payload: str, source: str) -> None:
    """Send an already-built payload (subscribe, handshake, non-chunked paths)."""
    await self._send_one(payload, source)

send_tellraw async 

send_tellraw(message: str, color: str, source: str, target: str = '@a') -> int

Send tellraw text and return the number of payloads actually transmitted.

Source code in src/mcbe_ws_sdk/delivery/outbound.py 
43
44
45
46
47
48
49
50
51
52
53
async def send_tellraw(
    self,
    message: str,
    color: str,
    source: str,
    target: str = "@a",
) -> int:
    """Send tellraw text and return the number of payloads actually transmitted."""
    payloads = self.flow.chunk_tellraw(message, color=color, target=target)
    await self.send_chunked(payloads, "tellraw", source)
    return len(payloads)

send_scriptevent async 

send_scriptevent(content: str, message_id: str = 'server:data', source: str = 'scriptevent') -> int

Send scriptevent text and return the number of payloads actually transmitted.

Source code in src/mcbe_ws_sdk/delivery/outbound.py 
55
56
57
58
59
60
61
62
63
64
async def send_scriptevent(
    self,
    content: str,
    message_id: str = "server:data",
    source: str = "scriptevent",
) -> int:
    """Send scriptevent text and return the number of payloads actually transmitted."""
    payloads = self.flow.chunk_scriptevent(content, message_id)
    await self.send_chunked(payloads, "scriptevent", source)
    return len(payloads)

send_raw_command async 

send_raw_command(command: str, source: str = 'raw_command', before_send: Callable[[str], None] | None = None) -> str

Send a raw command that must not be semantically split.

Raises FrameTooLargeError when the command exceeds the byte budget.

Source code in src/mcbe_ws_sdk/delivery/outbound.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
async def send_raw_command(
    self,
    command: str,
    source: str = "raw_command",
    before_send: Callable[[str], None] | None = None,
) -> str:
    """Send a raw command that must not be semantically split.

    Raises ``FrameTooLargeError`` when the command exceeds the byte budget.
    """
    payload = self.flow.chunk_raw_command(command)[0]
    request_id = _request_id_from_payload(payload)
    if before_send is not None:
        before_send(request_id)
    await self._send_one(payload, source)
    return request_id

send_outbound_text async 

send_outbound_text(message: OutboundText) -> int

Route an OutboundText message through the delivery adapter.

Source code in src/mcbe_ws_sdk/delivery/outbound.py 
83
84
85
86
87
88
89
90
91
async def send_outbound_text(self, message: OutboundText) -> int:
    """Route an OutboundText message through the delivery adapter."""
    source = f"outbound_text:{message.channel}"
    if message.delivery == "scriptevent":
        return await self.send_scriptevent(
            message.content, message_id=message.message_id, source=source
        )
    target = message.target or message.player_name or "@a"
    return await self.send_tellraw(message.content, color="", source=source, target=target)

send_system_notification async 

send_system_notification(message: SystemNotification) -> int

Route a SystemNotification through the delivery adapter.

Source code in src/mcbe_ws_sdk/delivery/outbound.py 
 93
 94
 95
 96
 97
 98
 99
100
101
async def send_system_notification(self, message: SystemNotification) -> int:
    """Route a SystemNotification through the delivery adapter."""
    colors = {"info": "§b", "warning": "§e", "error": "§c"}
    return await self.send_tellraw(
        message.message,
        color=colors[message.level],
        source="system_notification",
        target=message.player_name or "@a",
    )

send_chunked async 

send_chunked(payloads: list[str], delay_kind: str, source: str, *, delay: float | None = None) -> None

Send payloads with inter-chunk delays.

When delay is provided it overrides flow.chunk_delay_for(delay_kind) so protocol profiles can inject their own cadence (e.g. profile.response_chunk_delay).

Source code in src/mcbe_ws_sdk/delivery/outbound.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
async def send_chunked(
    self,
    payloads: list[str],
    delay_kind: str,
    source: str,
    *,
    delay: float | None = None,
) -> None:
    """Send ``payloads`` with inter-chunk delays.

    When ``delay`` is provided it overrides ``flow.chunk_delay_for(delay_kind)``
    so protocol profiles can inject their own cadence (e.g.
    ``profile.response_chunk_delay``).
    """
    chunk_delay = self.flow.chunk_delay_for(delay_kind) if delay is None else delay
    for idx, payload in enumerate(payloads):
        await self._send_one(
            payload,
            source=f"{source}_chunked" if len(payloads) > 1 else source,
        )
        if len(payloads) > 1 and idx < len(payloads) - 1:
            await asyncio.sleep(chunk_delay)

BridgeClosedError 

BridgeClosedError(request_id: str)

Bases: BridgeError

Raised when an add-on bridge is closed.

Source code in src/mcbe_ws_sdk/errors.py 
35
36
37
def __init__(self, request_id: str) -> None:
    super().__init__(f"Bridge connection closed: {request_id}")
    self.request_id = request_id

BridgeError

Bases: McbeWsSdkError

Base class for add-on bridge errors.

BridgeLimitError

Bases: BridgeError, ProtocolError

Raised when an add-on bridge protocol limit is exceeded.

BridgeTimeoutError 

BridgeTimeoutError(request_id: str)

Bases: BridgeError

Raised when an add-on bridge request times out.

Source code in src/mcbe_ws_sdk/errors.py 
27
28
29
def __init__(self, request_id: str) -> None:
    super().__init__(f"Bridge request timed out: {request_id}")
    self.request_id = request_id

ConfigurationError

Bases: McbeWsSdkError, ValueError

Raised when SDK settings are invalid.

FacadeLifecycleError

Bases: McbeWsSdkError, RuntimeError

Raised for invalid server facade lifecycle transitions.

FrameTooLargeError

Bases: ProtocolError

Raised when a frame cannot fit within its configured byte budget.

McbeWsSdkError

Bases: Exception

Base class for SDK errors.

ProtocolError

Bases: McbeWsSdkError, ValueError

Raised when a protocol contract is violated.

FlowControlMiddleware 

FlowControlMiddleware(settings: FlowControlSettings)

Unified flow control: split outbound long text into safe Minecraft commands.

Methods:

  • chunk_delay_for –

    Return inter-chunk delay in seconds for a chunking scenario.

  • chunk_tellraw –

    Split a long tellraw message into commandRequest JSON payload strings.

  • chunk_scriptevent –

    Split a long scriptevent payload into commandRequest JSON strings.

  • chunk_raw_command –

    Wrap a raw command as a one-element commandRequest JSON list.

Source code in src/mcbe_ws_sdk/flow/flow_control.py 
20
21
def __init__(self, settings: FlowControlSettings) -> None:
    self.settings = settings

chunk_delay_for 

chunk_delay_for(kind: str) -> float

Return inter-chunk delay in seconds for a chunking scenario.

kind is typically "tellraw" or "scriptevent". Unknown kinds yield 0.0 (no delay); the caller may treat that as an error.

Source code in src/mcbe_ws_sdk/flow/flow_control.py 
34
35
36
37
38
39
40
def chunk_delay_for(self, kind: str) -> float:
    """Return inter-chunk delay in seconds for a chunking scenario.

    ``kind`` is typically ``"tellraw"`` or ``"scriptevent"``. Unknown kinds
    yield ``0.0`` (no delay); the caller may treat that as an error.
    """
    return self.settings.chunk_delays.get(kind, 0.0)

chunk_tellraw 

chunk_tellraw(message: str, color: str = '§a', max_length: int | None = None, target: str = '@a') -> list[str]

Split a long tellraw message into commandRequest JSON payload strings.

Empty text returns []; the caller decides whether anything still must be sent. Each JSON commandLine text is at most max_length characters, and the final create_tellraw commandLine stays within the measured 461 B MCBE budget.

Source code in src/mcbe_ws_sdk/flow/flow_control.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
def chunk_tellraw(
    self,
    message: str,
    color: str = "§a",
    max_length: int | None = None,
    target: str = "@a",
) -> list[str]:
    """Split a long tellraw message into ``commandRequest`` JSON payload strings.

    Empty text returns ``[]``; the caller decides whether anything still must be
    sent. Each JSON ``commandLine`` text is at most ``max_length`` characters,
    and the final ``create_tellraw`` ``commandLine`` stays within the measured
    461 B MCBE budget.
    """
    if not message:
        return []

    max_len = self._get_max_length(max_length)
    text_parts = self._split_tellraw_text(
        message, color=color, target=target, max_length=max_len
    )

    payloads: list[str] = []
    for part in text_parts:
        command = MinecraftCommand.create_tellraw(part, color=color, target=target)
        payload = command.model_dump_json(exclude_none=True)
        self._assert_byte_safe(payload)
        payloads.append(payload)
    return payloads

chunk_scriptevent 

chunk_scriptevent(content: str, message_id: str = 'server:data', max_length: int | None = None) -> list[str]

Split a long scriptevent payload into commandRequest JSON strings.

Empty content returns []; the caller decides whether anything still must be sent. Each commandLine content segment is at most max_length characters, and total commandLine bytes stay within the measured 461 B MCBE budget.

Source code in src/mcbe_ws_sdk/flow/flow_control.py 
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def chunk_scriptevent(
    self,
    content: str,
    message_id: str = "server:data",
    max_length: int | None = None,
) -> list[str]:
    """Split a long scriptevent payload into ``commandRequest`` JSON strings.

    Empty content returns ``[]``; the caller decides whether anything still
    must be sent. Each ``commandLine`` content segment is at most
    ``max_length`` characters, and total ``commandLine`` bytes stay within
    the measured 461 B MCBE budget.
    """
    # Validate even empty content so invalid message_id is always rejected
    # before callers decide whether to send zero chunks.
    MinecraftCommand.create_scriptevent("", message_id)
    if not content:
        return []

    max_len = self._get_max_length(max_length)
    text_parts = self._split_by_command_fit(
        content,
        max_len,
        lambda part: MinecraftCommand.create_scriptevent(part, message_id).body.commandLine,
    )

    payloads: list[str] = []
    for part in text_parts:
        command = MinecraftCommand.create_scriptevent(part, message_id)
        payload = command.model_dump_json(exclude_none=True)
        self._assert_byte_safe(payload)
        payloads.append(payload)
    return payloads

chunk_raw_command 

chunk_raw_command(command: str) -> list[str]

Wrap a raw command as a one-element commandRequest JSON list.

Raw commands must not be truncated past the verb, or later fragments would be illegal. This method does not chunk: over-budget input raises FrameTooLargeError for the caller to handle.

Source code in src/mcbe_ws_sdk/flow/flow_control.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
def chunk_raw_command(self, command: str) -> list[str]:
    """Wrap a raw command as a one-element ``commandRequest`` JSON list.

    Raw commands must not be truncated past the verb, or later fragments would
    be illegal. This method **does not chunk**: over-budget input raises
    ``FrameTooLargeError`` for the caller to handle.
    """
    cmd = MinecraftCommand.create_raw(command)
    payload = cmd.model_dump_json(exclude_none=True)
    budget = self._byte_budget
    command_line_bytes = len(cmd.body.commandLine.encode("utf-8"))
    if command_line_bytes > budget:
        raise FrameTooLargeError(
            f"raw command too long in bytes "
            f"({command_line_bytes} > {budget}); "
            "cannot be safely chunked"
        )
    return [payload]

ConnectionHook

Bases: Protocol

Lifecycle + protocol hooks the host injects into the gateway.

Methods:

on_connected async 

on_connected(state: ConnectionState) -> None

Fired after the transport connection is established.

Source code in src/mcbe_ws_sdk/gateway/hook.py 
31
32
33
async def on_connected(self, state: ConnectionState) -> None:
    """Fired after the transport connection is established."""
    ...

on_disconnected async 

on_disconnected(state: ConnectionState) -> None

Fired on transport disconnect — host clears per-connection state here.

Source code in src/mcbe_ws_sdk/gateway/hook.py 
35
36
37
async def on_disconnected(self, state: ConnectionState) -> None:
    """Fired on transport disconnect — host clears per-connection state here."""
    ...

on_player_message async 

on_player_message(state: ConnectionState, player_event: PlayerMessageEvent, parsed: ParsedCommand | None = None) -> None

Inbound chat/scriptevent from a player.

parsed is the registry match for player_event.message when one exists; None means free-form chat (or no matching command prefix).

Source code in src/mcbe_ws_sdk/gateway/hook.py 
39
40
41
42
43
44
45
46
47
48
49
50
async def on_player_message(
    self,
    state: ConnectionState,
    player_event: PlayerMessageEvent,
    parsed: ParsedCommand | None = None,
) -> None:
    """Inbound chat/scriptevent from a player.

    ``parsed`` is the registry match for ``player_event.message`` when one
    exists; ``None`` means free-form chat (or no matching command prefix).
    """
    ...

on_ui_chat_reassembled async 

on_ui_chat_reassembled(state: ConnectionState, player_name: str, message: str) -> None

Fired when a fragmented UI_CHAT message is fully reassembled.

Source code in src/mcbe_ws_sdk/gateway/hook.py 
52
53
54
55
56
57
58
59
async def on_ui_chat_reassembled(
    self,
    state: ConnectionState,
    player_name: str,
    message: str,
) -> None:
    """Fired when a fragmented UI_CHAT message is fully reassembled."""
    ...

on_command_response async 

on_command_response(state: ConnectionState, response: MinecraftCommandResponse) -> None

Inbound commandResponse (resolves run_command futures).

Source code in src/mcbe_ws_sdk/gateway/hook.py 
61
62
63
64
65
66
67
async def on_command_response(
    self,
    state: ConnectionState,
    response: MinecraftCommandResponse,
) -> None:
    """Inbound ``commandResponse`` (resolves run_command futures)."""
    ...

on_error async 

on_error(state: ConnectionState, error: MinecraftErrorFrame) -> None

Inbound typed error frame.

Source code in src/mcbe_ws_sdk/gateway/hook.py 
69
70
71
async def on_error(self, state: ConnectionState, error: MinecraftErrorFrame) -> None:
    """Inbound typed ``error`` frame."""
    ...

ConnectionManager 

ConnectionManager(*, sink: ResponseSink | None = None, event_bus: EventBus | None = None, response_queue_maxsize: int = _DEFAULT_RESPONSE_QUEUE_MAXSIZE)

Owns active connections and their response-sender coroutines.

Construction takes injectable collaborators (sink / event bus) so the facade can supply host-backed variants; both fall back to gateway defaults.

Methods:

  • create_connection –

    Register a new connection and start its response-sender coroutine.

  • drop_connection –

    Cancel the sender coroutine, drop the connection, emit DISCONNECTED.

  • shutdown_all –

    Drop every connection (on server stop).

Source code in src/mcbe_ws_sdk/gateway/connection.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
def __init__(
    self,
    *,
    sink: ResponseSink | None = None,
    event_bus: EventBus | None = None,
    response_queue_maxsize: int = _DEFAULT_RESPONSE_QUEUE_MAXSIZE,
) -> None:
    if type(response_queue_maxsize) is not int or response_queue_maxsize <= 0:
        raise ValueError("response_queue_maxsize must be a positive integer")
    self._sink = sink or DefaultResponseSink()
    self._bus = event_bus or EventBus()
    self._response_queue_maxsize = response_queue_maxsize
    self._connections: dict[UUID, ConnectionState] = {}
    self._sender_tasks: dict[UUID, asyncio.Task[None]] = {}

create_connection async 

create_connection(*, send_payload: SendPayload, connection_id: UUID | None = None) -> ConnectionState

Register a new connection and start its response-sender coroutine.

The returned state carries a fresh bounded response_queue; the host posts response messages to it (prefer :func:enqueue_response so a full queue drops the oldest item instead of blocking). send_payload is the transport frame-send the sink's command routes ultimately deliver through.

Does not emit WsEventType.CONNECTED — that is the facade's job after handshake + subscribe succeed.

Source code in src/mcbe_ws_sdk/gateway/connection.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
async def create_connection(
    self,
    *,
    send_payload: SendPayload,
    connection_id: UUID | None = None,
) -> ConnectionState:
    """Register a new connection and start its response-sender coroutine.

    The returned state carries a fresh bounded ``response_queue``; the host
    posts response messages to it (prefer :func:`enqueue_response` so a full
    queue drops the oldest item instead of blocking). ``send_payload`` is the
    transport frame-send the sink's command routes ultimately deliver through.

    Does **not** emit ``WsEventType.CONNECTED`` — that is the facade's job
    after handshake + subscribe succeed.
    """
    state = ConnectionState(id=connection_id or uuid4(), send_payload=send_payload)
    state.response_queue = asyncio.Queue(maxsize=self._response_queue_maxsize)
    self._connections[state.id] = state
    self._sender_tasks[state.id] = asyncio.create_task(self._response_sender(state))
    logger.info("connection_created", connection_id=str(state.id))
    return state

drop_connection async 

drop_connection(connection_id: UUID) -> None

Cancel the sender coroutine, drop the connection, emit DISCONNECTED.

Source code in src/mcbe_ws_sdk/gateway/connection.py 
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
async def drop_connection(self, connection_id: UUID) -> None:
    """Cancel the sender coroutine, drop the connection, emit DISCONNECTED."""
    state = self._connections.pop(connection_id, None)
    if state is not None and state.response_queue is not None:
        discarded = state.response_queue.qsize()
        if discarded:
            logger.warning(
                "response_queue_discarded",
                connection_id=str(connection_id),
                discarded=discarded,
            )
    await self._cancel_sender(connection_id)
    if state is not None:
        await self._bus.emit(WsEventType.DISCONNECTED, state)
        logger.info("connection_dropped", connection_id=str(connection_id))

shutdown_all async 

shutdown_all() -> None

Drop every connection (on server stop).

Source code in src/mcbe_ws_sdk/gateway/connection.py 
185
186
187
188
async def shutdown_all(self) -> None:
    """Drop every connection (on server stop)."""
    for connection_id in list(self._connections):
        await self.drop_connection(connection_id)

ConnectionState dataclass 

ConnectionState(id: UUID = uuid4(), _player_name: str | None = None, send_payload: SendPayload | None = None, response_queue: Queue[object] | None = None)

Transport-agnostic connection identity.

Host-specific / transport wiring lives on the state but typed as opaque callables so the gateway never imports websockets itself: the facade or a host transport adapter sets send_payload. response_queue is the inbound message stream the response sender drains.

Attributes:

  • player_name (str | None) –

    Most-recent speaker convenience pointer only.

player_name property 

player_name: str | None

Most-recent speaker convenience pointer only.

.. deprecated:: Use :attr:PlayerMessageEvent.sender for authoritative player identity. This attribute is retained for backwards compatibility only and will emit a :class:DeprecationWarning on access.

DefaultResponseSink

Gateway default sink: logs metadata, delivers nothing to game.

on_outbound_text and on_system_notification log metadata (no player text, no command lines). A real host wires a :class:~mcbe_ws_sdk.delivery.outbound.McbeOutboundDelivery here.

dispatch remains as a non-protocol convenience for hosts/tests that want envelope-based routing; the manager never requires it.

Methods:

  • dispatch –

    Convenience router; not part of :class:ResponseSink.

dispatch async 

dispatch(state: ConnectionState, envelope: RouteEnvelope) -> None

Convenience router; not part of :class:ResponseSink.

Source code in src/mcbe_ws_sdk/gateway/sink.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
async def dispatch(self, state: ConnectionState, envelope: RouteEnvelope) -> None:
    """Convenience router; not part of :class:`ResponseSink`."""
    if envelope.kind is ResponseKind.OUTBOUND_TEXT:
        if not isinstance(envelope.payload, OutboundText):
            raise TypeError(
                "Expected OutboundText for OUTBOUND_TEXT kind, "
                f"got {type(envelope.payload).__name__}"
            )
        await self.on_outbound_text(state, envelope.payload)
        return
    if not isinstance(envelope.payload, SystemNotification):
        raise TypeError(
            "Expected SystemNotification for SYSTEM_NOTIFICATION kind, "
            f"got {type(envelope.payload).__name__}"
        )
    await self.on_system_notification(state, envelope.payload)

EventBus 

EventBus()

A typed in-process event bus keyed by :class:WsEventType.

Methods:

  • subscribe –

    Register handler and return a token for this registration.

  • unsubscribe –

    Remove exactly the registration identified by token.

  • handler_count –

    Return the number of live registrations for event.

  • emit –

    Await live handlers in subscription order.

Source code in src/mcbe_ws_sdk/gateway/events.py 
75
76
77
78
def __init__(self) -> None:
    self._subscribers: dict[WsEventType, dict[UUID, _Subscription]] = {
        event: {} for event in WsEventType
    }

subscribe 

subscribe(event: WsEventType, handler: Handler, *, weak: bool = True) -> SubscriptionToken

Register handler and return a token for this registration.

Source code in src/mcbe_ws_sdk/gateway/events.py 
80
81
82
83
84
85
86
87
88
89
90
def subscribe(
    self,
    event: WsEventType,
    handler: Handler,
    *,
    weak: bool = True,
) -> SubscriptionToken:
    """Register ``handler`` and return a token for this registration."""
    token = SubscriptionToken(event=event, id=uuid4())
    self._subscribers[event][token.id] = _Subscription.from_handler(token, handler, weak)
    return token

unsubscribe 

unsubscribe(token: SubscriptionToken) -> bool

Remove exactly the registration identified by token.

Source code in src/mcbe_ws_sdk/gateway/events.py 
92
93
94
def unsubscribe(self, token: SubscriptionToken) -> bool:
    """Remove exactly the registration identified by ``token``."""
    return self._subscribers[token.event].pop(token.id, None) is not None

handler_count 

handler_count(event: WsEventType) -> int

Return the number of live registrations for event.

Source code in src/mcbe_ws_sdk/gateway/events.py 
96
97
98
99
def handler_count(self, event: WsEventType) -> int:
    """Return the number of live registrations for ``event``."""
    self._prune_dead(event)
    return len(self._subscribers[event])

emit async 

emit(event: WsEventType, *args: Any, **kwargs: Any) -> None

Await live handlers in subscription order.

Each handler is isolated: an exception in one handler does not prevent subsequent handlers from being called.

Source code in src/mcbe_ws_sdk/gateway/events.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
async def emit(self, event: WsEventType, *args: Any, **kwargs: Any) -> None:
    """Await live handlers in subscription order.

    Each handler is isolated: an exception in one handler does not prevent
    subsequent handlers from being called.
    """
    subscribers = self._subscribers[event]
    for token_id in list(subscribers):
        subscription = subscribers.get(token_id)
        if subscription is None:
            continue
        handler = subscription.resolve()
        if handler is None:
            subscribers.pop(token_id, None)
            continue
        try:
            result = handler(*args, **kwargs)
            if inspect.isawaitable(result):
                await result
            elif result is not None:
                handler_name = getattr(handler, "__qualname__", repr(handler))
                raise TypeError(
                    f"Event handler {handler_name} for {event.value!r} must return None "
                    f"or an awaitable, got {type(result).__name__}"
                )
        except Exception:
            handler_name = getattr(handler, "__qualname__", repr(handler))
            structlog.get_logger(__name__).exception(
                "event_handler_failed",
                event_type=event.value,
                handler=handler_name,
            )

McbeServerFacade 

McbeServerFacade(*, settings: GatewaySettings | None = None, hook: ConnectionHook | None = None, sink: ResponseSink | None = None, addon: AddonBridgeService | None = None, registry: CommandRegistry | None = None, surface: MessageSurfaceConfig | None = None)

Owns the WS transport + connection/protocol machinery; host injects the rest.

The constructor takes ONLY keyword arguments and never builds a broker. Each None collapses to a gateway default so a host can stand up a working facade with McbeServerFacade() and override collaborators one at a time.

Methods:

  • run_lifetime –

    Bind a WebSocket server and serve until :meth:stop / task cancellation.

  • stop –

    Signal :meth:run_lifetime to unwind (idempotent).

Attributes:

Source code in src/mcbe_ws_sdk/gateway/server_facade.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def __init__(
    self,
    *,
    settings: GatewaySettings | None = None,
    hook: ConnectionHook | None = None,
    sink: ResponseSink | None = None,
    addon: AddonBridgeService | None = None,
    registry: CommandRegistry | None = None,
    surface: MessageSurfaceConfig | None = None,
) -> None:
    self._settings = settings if settings is not None else GatewaySettings()
    self._hook = hook if hook is not None else NoOpHook()
    self._sink = sink if sink is not None else DefaultResponseSink()
    self._registry = registry if registry is not None else CommandRegistry()

    self._handler = MinecraftProtocolHandler(self._registry, surface=surface)
    self._addon = addon if addon is not None else AddonBridgeService(self._settings.addon)
    self._addon.set_ui_chat_callback(self._on_ui_chat_reassembled)

    self._manager = ConnectionManager(
        sink=self._sink,
        event_bus=EventBus(),
        response_queue_maxsize=self._settings.websocket.response_queue_maxsize,
    )

    self._stopped = asyncio.Event()
    self._lifetime_started = False
    self._server: Any = None

manager property 

manager: ConnectionManager

The facade's owned connection manager.

handler property 

handler: MinecraftProtocolHandler

The facade's owned protocol handler (command parser + renderer).

addon property 

addon: AddonBridgeService

The facade's owned addon-bridge service instance.

run_lifetime async 

run_lifetime(host: str | None = None, port: int | None = None) -> None

Bind a WebSocket server and serve until :meth:stop / task cancellation.

host/port default to settings.websocket.host / port when None. Uses the websockets async-context-manager API (async with serve(...) as server), which works on both >=12 and the v13+ serve shape. Blocks on an internal asyncio.Event so an explicit stop() returns cleanly; the outer task being cancelled also unwinds into :meth:_graceful_shutdown.

Source code in src/mcbe_ws_sdk/gateway/server_facade.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
async def run_lifetime(
    self,
    host: str | None = None,
    port: int | None = None,
) -> None:
    """Bind a WebSocket server and serve until :meth:`stop` / task cancellation.

    ``host``/``port`` default to ``settings.websocket.host`` / ``port`` when
    ``None``. Uses the ``websockets`` async-context-manager API
    (``async with serve(...) as server``), which works on both ``>=12`` and
    the v13+ ``serve`` shape. Blocks on an internal ``asyncio.Event`` so an
    explicit ``stop()`` returns cleanly; the outer task being cancelled also
    unwinds into :meth:`_graceful_shutdown`.
    """
    if self._lifetime_started:
        raise FacadeLifecycleError("McbeServerFacade is single-use")
    self._lifetime_started = True
    serve_host = host if host is not None else self._settings.websocket.host
    serve_port = port if port is not None else self._settings.websocket.port
    transport = self._settings.websocket

    try:
        async with websockets.serve(
            self._on_connection,
            serve_host,
            serve_port,
            ping_interval=transport.ping_interval,
            ping_timeout=transport.ping_timeout,
            close_timeout=transport.close_timeout,
            max_size=transport.max_size,
            max_queue=transport.max_queue,
        ) as server:
            self._server = server
            logger.info(
                "facade_listening",
                host=serve_host,
                port=serve_port,
            )
            await self._stopped.wait()
    finally:
        try:
            await self._graceful_shutdown()
        finally:
            self._server = None

stop async 

stop() -> None

Signal :meth:run_lifetime to unwind (idempotent).

Source code in src/mcbe_ws_sdk/gateway/server_facade.py 
162
163
164
async def stop(self) -> None:
    """Signal :meth:`run_lifetime` to unwind (idempotent)."""
    self._stopped.set()

MessageSurfaceConfig dataclass 

MessageSurfaceConfig(welcome_message_template: str = '-----------\n已连接到 MCBE WebSocket 网关\n连接 ID: {connection_id}...\n-----------', error_prefix: str = '❌ 错误: ', info_prefix: str = 'ℹ ', success_prefix: str = '✅ ', error_color: str = '§c', info_color: str = '§b', success_color: str = '§a')

Presentation strings the protocol handler renders status messages with.

A frozen value object so a host can supply any wording/colors via config without the handler importing the host's settings model.

MinecraftProtocolHandler 

MinecraftProtocolHandler(command_registry: CommandRegistry, surface: MessageSurfaceConfig | None = None)

Constructs / parses MCBE connection-lifecycle protocol messages.

__init__ takes the collaborators it needs (a command registry and a presentation surface) and never imports anything host-specific.

Methods:

Source code in src/mcbe_ws_sdk/gateway/handler.py 
85
86
87
88
89
90
91
def __init__(
    self,
    command_registry: CommandRegistry,
    surface: MessageSurfaceConfig | None = None,
) -> None:
    self.command_registry = command_registry
    self.surface = surface or MessageSurfaceConfig()

create_subscribe_message staticmethod 

create_subscribe_message() -> str

Build the subscribe payload string for PlayerMessage events.

Source code in src/mcbe_ws_sdk/gateway/handler.py 
93
94
95
96
97
@staticmethod
def create_subscribe_message() -> str:
    """Build the ``subscribe`` payload string for PlayerMessage events."""
    subscribe = MinecraftSubscribe.player_message()
    return subscribe.model_dump_json(exclude_none=True)

create_welcome_message 

create_welcome_message(*, connection_id: str) -> str

Render the post-connect welcome banner (plain text; wrap a message).

Source code in src/mcbe_ws_sdk/gateway/handler.py 
 99
100
101
102
103
104
105
106
107
def create_welcome_message(
    self,
    *,
    connection_id: str,
) -> str:
    """Render the post-connect welcome banner (plain text; wrap a message)."""
    return self.surface.welcome_message_template.format(
        connection_id=connection_id[:8],
    )

parse_player_message staticmethod 

parse_player_message(data: dict[str, Any]) -> PlayerMessageEvent | None

Parse an inbound PlayerMessage event out of a WS frame body.

Source code in src/mcbe_ws_sdk/gateway/handler.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
@staticmethod
def parse_player_message(data: dict[str, Any]) -> PlayerMessageEvent | None:
    """Parse an inbound ``PlayerMessage`` event out of a WS frame body."""
    try:
        header = data.get("header", {})
        event_name = header.get("eventName")
        if event_name != "PlayerMessage":
            return None
        body = data.get("body", {})
        return PlayerMessageEvent.from_event_body(body)
    except Exception:
        logger.warning(
            "parse_player_message_error",
            error_type="protocol_parse_failed",
        )
        return None

parse_command 

parse_command(message: str) -> tuple[str | None, str]

Resolve message to (command_type, content) via the registry.

Source code in src/mcbe_ws_sdk/gateway/handler.py 
126
127
128
def parse_command(self, message: str) -> tuple[str | None, str]:
    """Resolve ``message`` to ``(command_type, content)`` via the registry."""
    return self.command_registry.resolve(message)

parse_typed_command 

parse_typed_command(message: str) -> ParsedCommand | None

Resolve message to a typed :class:ParsedCommand, or None.

Source code in src/mcbe_ws_sdk/gateway/handler.py 
130
131
132
def parse_typed_command(self, message: str) -> ParsedCommand | None:
    """Resolve ``message`` to a typed :class:`ParsedCommand`, or ``None``."""
    return self.command_registry.resolve_parsed(message)

get_help_text 

get_help_text() -> str

Render the in-game help listing from the command registry.

Source code in src/mcbe_ws_sdk/gateway/handler.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
def get_help_text(self) -> str:
    """Render the in-game help listing from the command registry."""
    lines = ["可用命令:"]
    for prefix, _cmd_type, _aliases in self.command_registry.list_all_commands():
        cmd_config = self.command_registry.get_command_config(prefix)
        if cmd_config is None:
            continue
        desc = cmd_config.description
        usage = cmd_config.usage
        if usage:
            lines.append(f"• {prefix} {usage} - {desc}")
        else:
            lines.append(f"• {prefix} - {desc}")
    return "\n".join(lines)

NoOpHook

Gateway default ConnectionHook — every hook is a no-op.

OutboundText dataclass 

OutboundText(content: str, channel: str = 'content', sequence: int = 0, delivery: DeliveryMode = 'tellraw', player_name: str | None = None, target: str | None = None, message_id: str = 'server:data', id: UUID = uuid4())

A text payload destined for a player (or all players) in the game.

ResponseKind

Bases: Enum

Message categories the response sender can route.

ResponseSink

Bases: Protocol

The two outbound delivery routes the response sender dispatches.

dispatch is intentionally not part of this protocol: the connection manager routes by envelope kind and calls the matching on_* method directly so a duck-typed host only needs these two hooks.

RouteEnvelope dataclass 

RouteEnvelope(kind: ResponseKind, payload: OutboundText | SystemNotification)

A response message the response sender routes to a sink method.

Methods:

  • from_message –

    Classify a host response into a :class:RouteEnvelope.

from_message classmethod 

from_message(message: object) -> RouteEnvelope

Classify a host response into a :class:RouteEnvelope.

Accepts only the gateway's own value objects (OutboundText, SystemNotification) by type. Anything else is rejected — the response loop should never silently drop an unroutable message.

Source code in src/mcbe_ws_sdk/gateway/sink.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
@classmethod
def from_message(cls, message: object) -> RouteEnvelope:
    """Classify a host response into a :class:`RouteEnvelope`.

    Accepts only the gateway's own value objects (OutboundText,
    SystemNotification) by type. Anything else is rejected — the response
    loop should never silently drop an unroutable message.
    """
    if isinstance(message, OutboundText):
        return cls(ResponseKind.OUTBOUND_TEXT, message)
    if isinstance(message, SystemNotification):
        return cls(ResponseKind.SYSTEM_NOTIFICATION, message)
    raise TypeError(f"Unroutable response message: {type(message).__name__}")

SubscriptionToken dataclass 

SubscriptionToken(event: WsEventType, id: UUID)

An opaque handle for one event-bus registration.

SystemNotification dataclass 

SystemNotification(level: NotificationLevel, message: str, player_name: str | None = None, id: UUID = uuid4())

A host/system status line surfaced to a player (or all players).

WebsocketTransportConfig dataclass 

WebsocketTransportConfig(host: str = '0.0.0.0', port: int = 8080, ping_interval: float | None = 30.0, ping_timeout: float | None = 15.0, close_timeout: float = 15.0, max_size: int | None = 10 * 1024 * 1024, max_queue: int | None = 32, response_queue_maxsize: int = 256)

Transport knobs for the facade's websockets.serve lifetime.

WsEventType

Bases: Enum

Events the connection lifetime and protocol handler can emit.

AddonBridgeProfile

Bases: Protocol

Wire-format contract for addon bridge interop.

PlayerMessageEvent

Bases: BaseModel

Parsed player chat/message event from the WebSocket stream.

Methods:

from_event_body classmethod 

from_event_body(body: dict[str, Any]) -> PlayerMessageEvent

Parse a PlayerMessageEvent from an event body dict.

Source code in src/mcbe_ws_sdk/protocol/minecraft.py 
195
196
197
198
199
200
201
202
203
@classmethod
def from_event_body(cls, body: dict[str, Any]) -> PlayerMessageEvent:
    """Parse a ``PlayerMessageEvent`` from an event body dict."""
    return cls(
        sender=body.get("sender", ""),
        message=body.get("message", ""),
        type=body.get("type"),
        receiver=body.get("receiver"),
    )

enqueue_response 

enqueue_response(state: ConnectionState, message: object) -> None

Put message onto state.response_queue, dropping the oldest if full.

No-ops when the connection has no queue. On overflow the oldest item is discarded and a warning is logged so a slow consumer cannot back-pressure the host forever.

Source code in src/mcbe_ws_sdk/gateway/connection.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def enqueue_response(state: ConnectionState, message: object) -> None:
    """Put ``message`` onto ``state.response_queue``, dropping the oldest if full.

    No-ops when the connection has no queue. On overflow the oldest item is
    discarded and a warning is logged so a slow consumer cannot back-pressure
    the host forever.
    """
    queue = state.response_queue
    if queue is None:
        return
    if queue.full():
        try:
            dropped = queue.get_nowait()
        except asyncio.QueueEmpty:
            dropped = None
        logger.warning(
            "response_queue_overflow_drop_oldest",
            connection_id=str(state.id),
            dropped_type=type(dropped).__name__ if dropped is not None else None,
            maxsize=queue.maxsize,
        )
    queue.put_nowait(message)

configure_logging 

configure_logging(level: LogLevel | str = 'INFO', *, colors: bool | None = None) -> None

Configure structlog + stdlib logging for compact console output.

Parameters

level: Root log level name (default INFO). colors: Force colour on/off. None (default) enables colour only when stdout is a TTY.

Source code in src/mcbe_ws_sdk/logging.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
def configure_logging(
    level: LogLevel | str = "INFO",
    *,
    colors: bool | None = None,
) -> None:
    """Configure structlog + stdlib logging for compact console output.

    Parameters
    ----------
    level:
        Root log level name (default ``INFO``).
    colors:
        Force colour on/off. ``None`` (default) enables colour only when stdout
        is a TTY.
    """
    level_name = str(level).upper()
    log_level = getattr(logging, level_name, None)
    if not isinstance(log_level, int):
        raise ValueError(f"unknown log level: {level!r}")

    use_colors = sys.stdout.isatty() if colors is None else colors

    shared_processors: list[structlog.types.Processor] = [
        structlog.contextvars.merge_contextvars,
        structlog.stdlib.add_log_level,
        structlog.processors.TimeStamper(fmt="%Y-%m-%d %H:%M:%S", utc=False),
        structlog.processors.StackInfoRenderer(),
        structlog.processors.format_exc_info,
        structlog.processors.UnicodeDecoder(),
    ]

    structlog.configure(
        processors=[
            *shared_processors,
            structlog.stdlib.ProcessorFormatter.wrap_for_formatter,
        ],
        wrapper_class=structlog.stdlib.BoundLogger,
        context_class=dict,
        logger_factory=structlog.stdlib.LoggerFactory(),
        cache_logger_on_first_use=True,
    )

    renderer: structlog.types.Processor
    if use_colors:
        renderer = structlog.dev.ConsoleRenderer(
            colors=True,
            pad_level=False,
            pad_event_to=0,
        )
    else:
        # Keep the same key=value layout when piping / redirecting; hosts that
        # want JSON can wire their own renderer.
        renderer = structlog.dev.ConsoleRenderer(
            colors=False,
            pad_level=False,
            pad_event_to=0,
        )

    formatter = structlog.stdlib.ProcessorFormatter(
        processor=renderer,
        foreign_pre_chain=shared_processors,
    )

    handler = logging.StreamHandler(sys.stdout)
    handler.setLevel(log_level)
    handler.setFormatter(formatter)

    root = logging.getLogger()
    root.handlers.clear()
    root.setLevel(log_level)
    root.addHandler(handler)

Gateway

mcbe_ws_sdk.gateway

MCBE WebSocket gateway SDK.

The gateway package owns the connection lifecycle, event bus, hook/sink protocols and the response routing machinery. The host application (the main repo) injects behaviour by implementing :class:~mcbe_ws_sdk.gateway.hook.ConnectionHook and :class:~mcbe_ws_sdk.gateway.sink.ResponseSink, then drives the stack through :class:~mcbe_ws_sdk.gateway.server_facade.McbeServerFacade.

Modules:

  • connection –

    Connection state + connection manager owned by the gateway.

  • events –

    WebSocket gateway event bus.

  • handler –

    MCBE WebSocket protocol handler.

  • hook –

    Connection lifecycle hook protocol.

  • messages –

    WebSocket gateway message value objects.

  • server_facade –

    Gateway entry point: the facade the host drives the SDK stack through.

  • sink –

    Response routing sink + RouteEnvelope value object + DefaultResponseSink.

Classes:

  • WebsocketTransportConfig –

    Transport knobs for the facade's websockets.serve lifetime.

  • ConnectionManager –

    Owns active connections and their response-sender coroutines.

  • ConnectionState –

    Transport-agnostic connection identity.

  • EventBus –

    A typed in-process event bus keyed by :class:WsEventType.

  • SubscriptionToken –

    An opaque handle for one event-bus registration.

  • WsEventType –

    Events the connection lifetime and protocol handler can emit.

  • MessageSurfaceConfig –

    Presentation strings the protocol handler renders status messages with.

  • MinecraftProtocolHandler –

    Constructs / parses MCBE connection-lifecycle protocol messages.

  • ConnectionHook –

    Lifecycle + protocol hooks the host injects into the gateway.

  • NoOpHook –

    Gateway default ConnectionHook — every hook is a no-op.

  • OutboundText –

    A text payload destined for a player (or all players) in the game.

  • SystemNotification –

    A host/system status line surfaced to a player (or all players).

  • McbeServerFacade –

    Owns the WS transport + connection/protocol machinery; host injects the rest.

  • DefaultResponseSink –

    Gateway default sink: logs metadata, delivers nothing to game.

  • ResponseKind –

    Message categories the response sender can route.

  • ResponseSink –

    The two outbound delivery routes the response sender dispatches.

  • RouteEnvelope –

    A response message the response sender routes to a sink method.

Functions:

  • enqueue_response –

    Put message onto state.response_queue, dropping the oldest if full.

WebsocketTransportConfig dataclass 

WebsocketTransportConfig(host: str = '0.0.0.0', port: int = 8080, ping_interval: float | None = 30.0, ping_timeout: float | None = 15.0, close_timeout: float = 15.0, max_size: int | None = 10 * 1024 * 1024, max_queue: int | None = 32, response_queue_maxsize: int = 256)

Transport knobs for the facade's websockets.serve lifetime.

ConnectionManager 

ConnectionManager(*, sink: ResponseSink | None = None, event_bus: EventBus | None = None, response_queue_maxsize: int = _DEFAULT_RESPONSE_QUEUE_MAXSIZE)

Owns active connections and their response-sender coroutines.

Construction takes injectable collaborators (sink / event bus) so the facade can supply host-backed variants; both fall back to gateway defaults.

Methods:

  • create_connection –

    Register a new connection and start its response-sender coroutine.

  • drop_connection –

    Cancel the sender coroutine, drop the connection, emit DISCONNECTED.

  • shutdown_all –

    Drop every connection (on server stop).

Source code in src/mcbe_ws_sdk/gateway/connection.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
def __init__(
    self,
    *,
    sink: ResponseSink | None = None,
    event_bus: EventBus | None = None,
    response_queue_maxsize: int = _DEFAULT_RESPONSE_QUEUE_MAXSIZE,
) -> None:
    if type(response_queue_maxsize) is not int or response_queue_maxsize <= 0:
        raise ValueError("response_queue_maxsize must be a positive integer")
    self._sink = sink or DefaultResponseSink()
    self._bus = event_bus or EventBus()
    self._response_queue_maxsize = response_queue_maxsize
    self._connections: dict[UUID, ConnectionState] = {}
    self._sender_tasks: dict[UUID, asyncio.Task[None]] = {}

create_connection async 

create_connection(*, send_payload: SendPayload, connection_id: UUID | None = None) -> ConnectionState

Register a new connection and start its response-sender coroutine.

The returned state carries a fresh bounded response_queue; the host posts response messages to it (prefer :func:enqueue_response so a full queue drops the oldest item instead of blocking). send_payload is the transport frame-send the sink's command routes ultimately deliver through.

Does not emit WsEventType.CONNECTED — that is the facade's job after handshake + subscribe succeed.

Source code in src/mcbe_ws_sdk/gateway/connection.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
async def create_connection(
    self,
    *,
    send_payload: SendPayload,
    connection_id: UUID | None = None,
) -> ConnectionState:
    """Register a new connection and start its response-sender coroutine.

    The returned state carries a fresh bounded ``response_queue``; the host
    posts response messages to it (prefer :func:`enqueue_response` so a full
    queue drops the oldest item instead of blocking). ``send_payload`` is the
    transport frame-send the sink's command routes ultimately deliver through.

    Does **not** emit ``WsEventType.CONNECTED`` — that is the facade's job
    after handshake + subscribe succeed.
    """
    state = ConnectionState(id=connection_id or uuid4(), send_payload=send_payload)
    state.response_queue = asyncio.Queue(maxsize=self._response_queue_maxsize)
    self._connections[state.id] = state
    self._sender_tasks[state.id] = asyncio.create_task(self._response_sender(state))
    logger.info("connection_created", connection_id=str(state.id))
    return state

drop_connection async 

drop_connection(connection_id: UUID) -> None

Cancel the sender coroutine, drop the connection, emit DISCONNECTED.

Source code in src/mcbe_ws_sdk/gateway/connection.py 
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
async def drop_connection(self, connection_id: UUID) -> None:
    """Cancel the sender coroutine, drop the connection, emit DISCONNECTED."""
    state = self._connections.pop(connection_id, None)
    if state is not None and state.response_queue is not None:
        discarded = state.response_queue.qsize()
        if discarded:
            logger.warning(
                "response_queue_discarded",
                connection_id=str(connection_id),
                discarded=discarded,
            )
    await self._cancel_sender(connection_id)
    if state is not None:
        await self._bus.emit(WsEventType.DISCONNECTED, state)
        logger.info("connection_dropped", connection_id=str(connection_id))

shutdown_all async 

shutdown_all() -> None

Drop every connection (on server stop).

Source code in src/mcbe_ws_sdk/gateway/connection.py 
185
186
187
188
async def shutdown_all(self) -> None:
    """Drop every connection (on server stop)."""
    for connection_id in list(self._connections):
        await self.drop_connection(connection_id)

ConnectionState dataclass 

ConnectionState(id: UUID = uuid4(), _player_name: str | None = None, send_payload: SendPayload | None = None, response_queue: Queue[object] | None = None)

Transport-agnostic connection identity.

Host-specific / transport wiring lives on the state but typed as opaque callables so the gateway never imports websockets itself: the facade or a host transport adapter sets send_payload. response_queue is the inbound message stream the response sender drains.

Attributes:

  • player_name (str | None) –

    Most-recent speaker convenience pointer only.

player_name property 

player_name: str | None

Most-recent speaker convenience pointer only.

.. deprecated:: Use :attr:PlayerMessageEvent.sender for authoritative player identity. This attribute is retained for backwards compatibility only and will emit a :class:DeprecationWarning on access.

EventBus 

EventBus()

A typed in-process event bus keyed by :class:WsEventType.

Methods:

  • subscribe –

    Register handler and return a token for this registration.

  • unsubscribe –

    Remove exactly the registration identified by token.

  • handler_count –

    Return the number of live registrations for event.

  • emit –

    Await live handlers in subscription order.

Source code in src/mcbe_ws_sdk/gateway/events.py 
75
76
77
78
def __init__(self) -> None:
    self._subscribers: dict[WsEventType, dict[UUID, _Subscription]] = {
        event: {} for event in WsEventType
    }

subscribe 

subscribe(event: WsEventType, handler: Handler, *, weak: bool = True) -> SubscriptionToken

Register handler and return a token for this registration.

Source code in src/mcbe_ws_sdk/gateway/events.py 
80
81
82
83
84
85
86
87
88
89
90
def subscribe(
    self,
    event: WsEventType,
    handler: Handler,
    *,
    weak: bool = True,
) -> SubscriptionToken:
    """Register ``handler`` and return a token for this registration."""
    token = SubscriptionToken(event=event, id=uuid4())
    self._subscribers[event][token.id] = _Subscription.from_handler(token, handler, weak)
    return token

unsubscribe 

unsubscribe(token: SubscriptionToken) -> bool

Remove exactly the registration identified by token.

Source code in src/mcbe_ws_sdk/gateway/events.py 
92
93
94
def unsubscribe(self, token: SubscriptionToken) -> bool:
    """Remove exactly the registration identified by ``token``."""
    return self._subscribers[token.event].pop(token.id, None) is not None

handler_count 

handler_count(event: WsEventType) -> int

Return the number of live registrations for event.

Source code in src/mcbe_ws_sdk/gateway/events.py 
96
97
98
99
def handler_count(self, event: WsEventType) -> int:
    """Return the number of live registrations for ``event``."""
    self._prune_dead(event)
    return len(self._subscribers[event])

emit async 

emit(event: WsEventType, *args: Any, **kwargs: Any) -> None

Await live handlers in subscription order.

Each handler is isolated: an exception in one handler does not prevent subsequent handlers from being called.

Source code in src/mcbe_ws_sdk/gateway/events.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
async def emit(self, event: WsEventType, *args: Any, **kwargs: Any) -> None:
    """Await live handlers in subscription order.

    Each handler is isolated: an exception in one handler does not prevent
    subsequent handlers from being called.
    """
    subscribers = self._subscribers[event]
    for token_id in list(subscribers):
        subscription = subscribers.get(token_id)
        if subscription is None:
            continue
        handler = subscription.resolve()
        if handler is None:
            subscribers.pop(token_id, None)
            continue
        try:
            result = handler(*args, **kwargs)
            if inspect.isawaitable(result):
                await result
            elif result is not None:
                handler_name = getattr(handler, "__qualname__", repr(handler))
                raise TypeError(
                    f"Event handler {handler_name} for {event.value!r} must return None "
                    f"or an awaitable, got {type(result).__name__}"
                )
        except Exception:
            handler_name = getattr(handler, "__qualname__", repr(handler))
            structlog.get_logger(__name__).exception(
                "event_handler_failed",
                event_type=event.value,
                handler=handler_name,
            )

SubscriptionToken dataclass 

SubscriptionToken(event: WsEventType, id: UUID)

An opaque handle for one event-bus registration.

WsEventType

Bases: Enum

Events the connection lifetime and protocol handler can emit.

MessageSurfaceConfig dataclass 

MessageSurfaceConfig(welcome_message_template: str = '-----------\n已连接到 MCBE WebSocket 网关\n连接 ID: {connection_id}...\n-----------', error_prefix: str = '❌ 错误: ', info_prefix: str = 'ℹ ', success_prefix: str = '✅ ', error_color: str = '§c', info_color: str = '§b', success_color: str = '§a')

Presentation strings the protocol handler renders status messages with.

A frozen value object so a host can supply any wording/colors via config without the handler importing the host's settings model.

MinecraftProtocolHandler 

MinecraftProtocolHandler(command_registry: CommandRegistry, surface: MessageSurfaceConfig | None = None)

Constructs / parses MCBE connection-lifecycle protocol messages.

__init__ takes the collaborators it needs (a command registry and a presentation surface) and never imports anything host-specific.

Methods:

Source code in src/mcbe_ws_sdk/gateway/handler.py 
85
86
87
88
89
90
91
def __init__(
    self,
    command_registry: CommandRegistry,
    surface: MessageSurfaceConfig | None = None,
) -> None:
    self.command_registry = command_registry
    self.surface = surface or MessageSurfaceConfig()

create_subscribe_message staticmethod 

create_subscribe_message() -> str

Build the subscribe payload string for PlayerMessage events.

Source code in src/mcbe_ws_sdk/gateway/handler.py 
93
94
95
96
97
@staticmethod
def create_subscribe_message() -> str:
    """Build the ``subscribe`` payload string for PlayerMessage events."""
    subscribe = MinecraftSubscribe.player_message()
    return subscribe.model_dump_json(exclude_none=True)

create_welcome_message 

create_welcome_message(*, connection_id: str) -> str

Render the post-connect welcome banner (plain text; wrap a message).

Source code in src/mcbe_ws_sdk/gateway/handler.py 
 99
100
101
102
103
104
105
106
107
def create_welcome_message(
    self,
    *,
    connection_id: str,
) -> str:
    """Render the post-connect welcome banner (plain text; wrap a message)."""
    return self.surface.welcome_message_template.format(
        connection_id=connection_id[:8],
    )

parse_player_message staticmethod 

parse_player_message(data: dict[str, Any]) -> PlayerMessageEvent | None

Parse an inbound PlayerMessage event out of a WS frame body.

Source code in src/mcbe_ws_sdk/gateway/handler.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
@staticmethod
def parse_player_message(data: dict[str, Any]) -> PlayerMessageEvent | None:
    """Parse an inbound ``PlayerMessage`` event out of a WS frame body."""
    try:
        header = data.get("header", {})
        event_name = header.get("eventName")
        if event_name != "PlayerMessage":
            return None
        body = data.get("body", {})
        return PlayerMessageEvent.from_event_body(body)
    except Exception:
        logger.warning(
            "parse_player_message_error",
            error_type="protocol_parse_failed",
        )
        return None

parse_command 

parse_command(message: str) -> tuple[str | None, str]

Resolve message to (command_type, content) via the registry.

Source code in src/mcbe_ws_sdk/gateway/handler.py 
126
127
128
def parse_command(self, message: str) -> tuple[str | None, str]:
    """Resolve ``message`` to ``(command_type, content)`` via the registry."""
    return self.command_registry.resolve(message)

parse_typed_command 

parse_typed_command(message: str) -> ParsedCommand | None

Resolve message to a typed :class:ParsedCommand, or None.

Source code in src/mcbe_ws_sdk/gateway/handler.py 
130
131
132
def parse_typed_command(self, message: str) -> ParsedCommand | None:
    """Resolve ``message`` to a typed :class:`ParsedCommand`, or ``None``."""
    return self.command_registry.resolve_parsed(message)

get_help_text 

get_help_text() -> str

Render the in-game help listing from the command registry.

Source code in src/mcbe_ws_sdk/gateway/handler.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
def get_help_text(self) -> str:
    """Render the in-game help listing from the command registry."""
    lines = ["可用命令:"]
    for prefix, _cmd_type, _aliases in self.command_registry.list_all_commands():
        cmd_config = self.command_registry.get_command_config(prefix)
        if cmd_config is None:
            continue
        desc = cmd_config.description
        usage = cmd_config.usage
        if usage:
            lines.append(f"• {prefix} {usage} - {desc}")
        else:
            lines.append(f"• {prefix} - {desc}")
    return "\n".join(lines)

ConnectionHook

Bases: Protocol

Lifecycle + protocol hooks the host injects into the gateway.

Methods:

on_connected async 

on_connected(state: ConnectionState) -> None

Fired after the transport connection is established.

Source code in src/mcbe_ws_sdk/gateway/hook.py 
31
32
33
async def on_connected(self, state: ConnectionState) -> None:
    """Fired after the transport connection is established."""
    ...

on_disconnected async 

on_disconnected(state: ConnectionState) -> None

Fired on transport disconnect — host clears per-connection state here.

Source code in src/mcbe_ws_sdk/gateway/hook.py 
35
36
37
async def on_disconnected(self, state: ConnectionState) -> None:
    """Fired on transport disconnect — host clears per-connection state here."""
    ...

on_player_message async 

on_player_message(state: ConnectionState, player_event: PlayerMessageEvent, parsed: ParsedCommand | None = None) -> None

Inbound chat/scriptevent from a player.

parsed is the registry match for player_event.message when one exists; None means free-form chat (or no matching command prefix).

Source code in src/mcbe_ws_sdk/gateway/hook.py 
39
40
41
42
43
44
45
46
47
48
49
50
async def on_player_message(
    self,
    state: ConnectionState,
    player_event: PlayerMessageEvent,
    parsed: ParsedCommand | None = None,
) -> None:
    """Inbound chat/scriptevent from a player.

    ``parsed`` is the registry match for ``player_event.message`` when one
    exists; ``None`` means free-form chat (or no matching command prefix).
    """
    ...

on_ui_chat_reassembled async 

on_ui_chat_reassembled(state: ConnectionState, player_name: str, message: str) -> None

Fired when a fragmented UI_CHAT message is fully reassembled.

Source code in src/mcbe_ws_sdk/gateway/hook.py 
52
53
54
55
56
57
58
59
async def on_ui_chat_reassembled(
    self,
    state: ConnectionState,
    player_name: str,
    message: str,
) -> None:
    """Fired when a fragmented UI_CHAT message is fully reassembled."""
    ...

on_command_response async 

on_command_response(state: ConnectionState, response: MinecraftCommandResponse) -> None

Inbound commandResponse (resolves run_command futures).

Source code in src/mcbe_ws_sdk/gateway/hook.py 
61
62
63
64
65
66
67
async def on_command_response(
    self,
    state: ConnectionState,
    response: MinecraftCommandResponse,
) -> None:
    """Inbound ``commandResponse`` (resolves run_command futures)."""
    ...

on_error async 

on_error(state: ConnectionState, error: MinecraftErrorFrame) -> None

Inbound typed error frame.

Source code in src/mcbe_ws_sdk/gateway/hook.py 
69
70
71
async def on_error(self, state: ConnectionState, error: MinecraftErrorFrame) -> None:
    """Inbound typed ``error`` frame."""
    ...

NoOpHook

Gateway default ConnectionHook — every hook is a no-op.

OutboundText dataclass 

OutboundText(content: str, channel: str = 'content', sequence: int = 0, delivery: DeliveryMode = 'tellraw', player_name: str | None = None, target: str | None = None, message_id: str = 'server:data', id: UUID = uuid4())

A text payload destined for a player (or all players) in the game.

SystemNotification dataclass 

SystemNotification(level: NotificationLevel, message: str, player_name: str | None = None, id: UUID = uuid4())

A host/system status line surfaced to a player (or all players).

McbeServerFacade 

McbeServerFacade(*, settings: GatewaySettings | None = None, hook: ConnectionHook | None = None, sink: ResponseSink | None = None, addon: AddonBridgeService | None = None, registry: CommandRegistry | None = None, surface: MessageSurfaceConfig | None = None)

Owns the WS transport + connection/protocol machinery; host injects the rest.

The constructor takes ONLY keyword arguments and never builds a broker. Each None collapses to a gateway default so a host can stand up a working facade with McbeServerFacade() and override collaborators one at a time.

Methods:

  • run_lifetime –

    Bind a WebSocket server and serve until :meth:stop / task cancellation.

  • stop –

    Signal :meth:run_lifetime to unwind (idempotent).

Attributes:

Source code in src/mcbe_ws_sdk/gateway/server_facade.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def __init__(
    self,
    *,
    settings: GatewaySettings | None = None,
    hook: ConnectionHook | None = None,
    sink: ResponseSink | None = None,
    addon: AddonBridgeService | None = None,
    registry: CommandRegistry | None = None,
    surface: MessageSurfaceConfig | None = None,
) -> None:
    self._settings = settings if settings is not None else GatewaySettings()
    self._hook = hook if hook is not None else NoOpHook()
    self._sink = sink if sink is not None else DefaultResponseSink()
    self._registry = registry if registry is not None else CommandRegistry()

    self._handler = MinecraftProtocolHandler(self._registry, surface=surface)
    self._addon = addon if addon is not None else AddonBridgeService(self._settings.addon)
    self._addon.set_ui_chat_callback(self._on_ui_chat_reassembled)

    self._manager = ConnectionManager(
        sink=self._sink,
        event_bus=EventBus(),
        response_queue_maxsize=self._settings.websocket.response_queue_maxsize,
    )

    self._stopped = asyncio.Event()
    self._lifetime_started = False
    self._server: Any = None

manager property 

manager: ConnectionManager

The facade's owned connection manager.

handler property 

handler: MinecraftProtocolHandler

The facade's owned protocol handler (command parser + renderer).

addon property 

addon: AddonBridgeService

The facade's owned addon-bridge service instance.

run_lifetime async 

run_lifetime(host: str | None = None, port: int | None = None) -> None

Bind a WebSocket server and serve until :meth:stop / task cancellation.

host/port default to settings.websocket.host / port when None. Uses the websockets async-context-manager API (async with serve(...) as server), which works on both >=12 and the v13+ serve shape. Blocks on an internal asyncio.Event so an explicit stop() returns cleanly; the outer task being cancelled also unwinds into :meth:_graceful_shutdown.

Source code in src/mcbe_ws_sdk/gateway/server_facade.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
async def run_lifetime(
    self,
    host: str | None = None,
    port: int | None = None,
) -> None:
    """Bind a WebSocket server and serve until :meth:`stop` / task cancellation.

    ``host``/``port`` default to ``settings.websocket.host`` / ``port`` when
    ``None``. Uses the ``websockets`` async-context-manager API
    (``async with serve(...) as server``), which works on both ``>=12`` and
    the v13+ ``serve`` shape. Blocks on an internal ``asyncio.Event`` so an
    explicit ``stop()`` returns cleanly; the outer task being cancelled also
    unwinds into :meth:`_graceful_shutdown`.
    """
    if self._lifetime_started:
        raise FacadeLifecycleError("McbeServerFacade is single-use")
    self._lifetime_started = True
    serve_host = host if host is not None else self._settings.websocket.host
    serve_port = port if port is not None else self._settings.websocket.port
    transport = self._settings.websocket

    try:
        async with websockets.serve(
            self._on_connection,
            serve_host,
            serve_port,
            ping_interval=transport.ping_interval,
            ping_timeout=transport.ping_timeout,
            close_timeout=transport.close_timeout,
            max_size=transport.max_size,
            max_queue=transport.max_queue,
        ) as server:
            self._server = server
            logger.info(
                "facade_listening",
                host=serve_host,
                port=serve_port,
            )
            await self._stopped.wait()
    finally:
        try:
            await self._graceful_shutdown()
        finally:
            self._server = None

stop async 

stop() -> None

Signal :meth:run_lifetime to unwind (idempotent).

Source code in src/mcbe_ws_sdk/gateway/server_facade.py 
162
163
164
async def stop(self) -> None:
    """Signal :meth:`run_lifetime` to unwind (idempotent)."""
    self._stopped.set()

DefaultResponseSink

Gateway default sink: logs metadata, delivers nothing to game.

on_outbound_text and on_system_notification log metadata (no player text, no command lines). A real host wires a :class:~mcbe_ws_sdk.delivery.outbound.McbeOutboundDelivery here.

dispatch remains as a non-protocol convenience for hosts/tests that want envelope-based routing; the manager never requires it.

Methods:

  • dispatch –

    Convenience router; not part of :class:ResponseSink.

dispatch async 

dispatch(state: ConnectionState, envelope: RouteEnvelope) -> None

Convenience router; not part of :class:ResponseSink.

Source code in src/mcbe_ws_sdk/gateway/sink.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
async def dispatch(self, state: ConnectionState, envelope: RouteEnvelope) -> None:
    """Convenience router; not part of :class:`ResponseSink`."""
    if envelope.kind is ResponseKind.OUTBOUND_TEXT:
        if not isinstance(envelope.payload, OutboundText):
            raise TypeError(
                "Expected OutboundText for OUTBOUND_TEXT kind, "
                f"got {type(envelope.payload).__name__}"
            )
        await self.on_outbound_text(state, envelope.payload)
        return
    if not isinstance(envelope.payload, SystemNotification):
        raise TypeError(
            "Expected SystemNotification for SYSTEM_NOTIFICATION kind, "
            f"got {type(envelope.payload).__name__}"
        )
    await self.on_system_notification(state, envelope.payload)

ResponseKind

Bases: Enum

Message categories the response sender can route.

ResponseSink

Bases: Protocol

The two outbound delivery routes the response sender dispatches.

dispatch is intentionally not part of this protocol: the connection manager routes by envelope kind and calls the matching on_* method directly so a duck-typed host only needs these two hooks.

RouteEnvelope dataclass 

RouteEnvelope(kind: ResponseKind, payload: OutboundText | SystemNotification)

A response message the response sender routes to a sink method.

Methods:

  • from_message –

    Classify a host response into a :class:RouteEnvelope.

from_message classmethod 

from_message(message: object) -> RouteEnvelope

Classify a host response into a :class:RouteEnvelope.

Accepts only the gateway's own value objects (OutboundText, SystemNotification) by type. Anything else is rejected — the response loop should never silently drop an unroutable message.

Source code in src/mcbe_ws_sdk/gateway/sink.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
@classmethod
def from_message(cls, message: object) -> RouteEnvelope:
    """Classify a host response into a :class:`RouteEnvelope`.

    Accepts only the gateway's own value objects (OutboundText,
    SystemNotification) by type. Anything else is rejected — the response
    loop should never silently drop an unroutable message.
    """
    if isinstance(message, OutboundText):
        return cls(ResponseKind.OUTBOUND_TEXT, message)
    if isinstance(message, SystemNotification):
        return cls(ResponseKind.SYSTEM_NOTIFICATION, message)
    raise TypeError(f"Unroutable response message: {type(message).__name__}")

enqueue_response 

enqueue_response(state: ConnectionState, message: object) -> None

Put message onto state.response_queue, dropping the oldest if full.

No-ops when the connection has no queue. On overflow the oldest item is discarded and a warning is logged so a slow consumer cannot back-pressure the host forever.

Source code in src/mcbe_ws_sdk/gateway/connection.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def enqueue_response(state: ConnectionState, message: object) -> None:
    """Put ``message`` onto ``state.response_queue``, dropping the oldest if full.

    No-ops when the connection has no queue. On overflow the oldest item is
    discarded and a warning is logged so a slow consumer cannot back-pressure
    the host forever.
    """
    queue = state.response_queue
    if queue is None:
        return
    if queue.full():
        try:
            dropped = queue.get_nowait()
        except asyncio.QueueEmpty:
            dropped = None
        logger.warning(
            "response_queue_overflow_drop_oldest",
            connection_id=str(state.id),
            dropped_type=type(dropped).__name__ if dropped is not None else None,
            maxsize=queue.maxsize,
        )
    queue.put_nowait(message)

connection

Connection state + connection manager owned by the gateway.

Extends the minimal :class:ConnectionState foundation with the lifecycle the host drives through the facade: per-connection response queues, an outbound send_payload callable (the transport's frame send, e.g. websocket.send), and a :class:ConnectionManager that owns the response-sender coroutine and emits DISCONNECTED on the :class:~mcbe_ws_sdk.gateway.events.EventBus.

CONNECTED is intentionally not emitted here — the facade emits it after a successful handshake + subscribe, at the same moment as hook.on_connected.

The response-sender loop never builds Minecraft commands itself. It classifies each queued message with :meth:~mcbe_ws_sdk.gateway.sink.RouteEnvelope.from_message and routes it inline to the sink's on_* methods, pushing the application-specific mapping entirely onto the host's HostSink.

Classes:

Functions:

  • enqueue_response –

    Put message onto state.response_queue, dropping the oldest if full.

ConnectionState dataclass 

ConnectionState(id: UUID = uuid4(), _player_name: str | None = None, send_payload: SendPayload | None = None, response_queue: Queue[object] | None = None)

Transport-agnostic connection identity.

Host-specific / transport wiring lives on the state but typed as opaque callables so the gateway never imports websockets itself: the facade or a host transport adapter sets send_payload. response_queue is the inbound message stream the response sender drains.

Attributes:

  • player_name (str | None) –

    Most-recent speaker convenience pointer only.

player_name property 
player_name: str | None

Most-recent speaker convenience pointer only.

.. deprecated:: Use :attr:PlayerMessageEvent.sender for authoritative player identity. This attribute is retained for backwards compatibility only and will emit a :class:DeprecationWarning on access.

ConnectionManager 

ConnectionManager(*, sink: ResponseSink | None = None, event_bus: EventBus | None = None, response_queue_maxsize: int = _DEFAULT_RESPONSE_QUEUE_MAXSIZE)

Owns active connections and their response-sender coroutines.

Construction takes injectable collaborators (sink / event bus) so the facade can supply host-backed variants; both fall back to gateway defaults.

Methods:

  • create_connection –

    Register a new connection and start its response-sender coroutine.

  • drop_connection –

    Cancel the sender coroutine, drop the connection, emit DISCONNECTED.

  • shutdown_all –

    Drop every connection (on server stop).

Source code in src/mcbe_ws_sdk/gateway/connection.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
def __init__(
    self,
    *,
    sink: ResponseSink | None = None,
    event_bus: EventBus | None = None,
    response_queue_maxsize: int = _DEFAULT_RESPONSE_QUEUE_MAXSIZE,
) -> None:
    if type(response_queue_maxsize) is not int or response_queue_maxsize <= 0:
        raise ValueError("response_queue_maxsize must be a positive integer")
    self._sink = sink or DefaultResponseSink()
    self._bus = event_bus or EventBus()
    self._response_queue_maxsize = response_queue_maxsize
    self._connections: dict[UUID, ConnectionState] = {}
    self._sender_tasks: dict[UUID, asyncio.Task[None]] = {}
create_connection async 
create_connection(*, send_payload: SendPayload, connection_id: UUID | None = None) -> ConnectionState

Register a new connection and start its response-sender coroutine.

The returned state carries a fresh bounded response_queue; the host posts response messages to it (prefer :func:enqueue_response so a full queue drops the oldest item instead of blocking). send_payload is the transport frame-send the sink's command routes ultimately deliver through.

Does not emit WsEventType.CONNECTED — that is the facade's job after handshake + subscribe succeed.

Source code in src/mcbe_ws_sdk/gateway/connection.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
async def create_connection(
    self,
    *,
    send_payload: SendPayload,
    connection_id: UUID | None = None,
) -> ConnectionState:
    """Register a new connection and start its response-sender coroutine.

    The returned state carries a fresh bounded ``response_queue``; the host
    posts response messages to it (prefer :func:`enqueue_response` so a full
    queue drops the oldest item instead of blocking). ``send_payload`` is the
    transport frame-send the sink's command routes ultimately deliver through.

    Does **not** emit ``WsEventType.CONNECTED`` — that is the facade's job
    after handshake + subscribe succeed.
    """
    state = ConnectionState(id=connection_id or uuid4(), send_payload=send_payload)
    state.response_queue = asyncio.Queue(maxsize=self._response_queue_maxsize)
    self._connections[state.id] = state
    self._sender_tasks[state.id] = asyncio.create_task(self._response_sender(state))
    logger.info("connection_created", connection_id=str(state.id))
    return state
drop_connection async 
drop_connection(connection_id: UUID) -> None

Cancel the sender coroutine, drop the connection, emit DISCONNECTED.

Source code in src/mcbe_ws_sdk/gateway/connection.py 
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
async def drop_connection(self, connection_id: UUID) -> None:
    """Cancel the sender coroutine, drop the connection, emit DISCONNECTED."""
    state = self._connections.pop(connection_id, None)
    if state is not None and state.response_queue is not None:
        discarded = state.response_queue.qsize()
        if discarded:
            logger.warning(
                "response_queue_discarded",
                connection_id=str(connection_id),
                discarded=discarded,
            )
    await self._cancel_sender(connection_id)
    if state is not None:
        await self._bus.emit(WsEventType.DISCONNECTED, state)
        logger.info("connection_dropped", connection_id=str(connection_id))
shutdown_all async 
shutdown_all() -> None

Drop every connection (on server stop).

Source code in src/mcbe_ws_sdk/gateway/connection.py 
185
186
187
188
async def shutdown_all(self) -> None:
    """Drop every connection (on server stop)."""
    for connection_id in list(self._connections):
        await self.drop_connection(connection_id)

enqueue_response 

enqueue_response(state: ConnectionState, message: object) -> None

Put message onto state.response_queue, dropping the oldest if full.

No-ops when the connection has no queue. On overflow the oldest item is discarded and a warning is logged so a slow consumer cannot back-pressure the host forever.

Source code in src/mcbe_ws_sdk/gateway/connection.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def enqueue_response(state: ConnectionState, message: object) -> None:
    """Put ``message`` onto ``state.response_queue``, dropping the oldest if full.

    No-ops when the connection has no queue. On overflow the oldest item is
    discarded and a warning is logged so a slow consumer cannot back-pressure
    the host forever.
    """
    queue = state.response_queue
    if queue is None:
        return
    if queue.full():
        try:
            dropped = queue.get_nowait()
        except asyncio.QueueEmpty:
            dropped = None
        logger.warning(
            "response_queue_overflow_drop_oldest",
            connection_id=str(state.id),
            dropped_type=type(dropped).__name__ if dropped is not None else None,
            maxsize=queue.maxsize,
        )
    queue.put_nowait(message)

events

WebSocket gateway event bus.

Classes:

  • WsEventType –

    Events the connection lifetime and protocol handler can emit.

  • SubscriptionToken –

    An opaque handle for one event-bus registration.

  • EventBus –

    A typed in-process event bus keyed by :class:WsEventType.

WsEventType

Bases: Enum

Events the connection lifetime and protocol handler can emit.

SubscriptionToken dataclass 

SubscriptionToken(event: WsEventType, id: UUID)

An opaque handle for one event-bus registration.

EventBus 

EventBus()

A typed in-process event bus keyed by :class:WsEventType.

Methods:

  • subscribe –

    Register handler and return a token for this registration.

  • unsubscribe –

    Remove exactly the registration identified by token.

  • handler_count –

    Return the number of live registrations for event.

  • emit –

    Await live handlers in subscription order.

Source code in src/mcbe_ws_sdk/gateway/events.py 
75
76
77
78
def __init__(self) -> None:
    self._subscribers: dict[WsEventType, dict[UUID, _Subscription]] = {
        event: {} for event in WsEventType
    }
subscribe 
subscribe(event: WsEventType, handler: Handler, *, weak: bool = True) -> SubscriptionToken

Register handler and return a token for this registration.

Source code in src/mcbe_ws_sdk/gateway/events.py 
80
81
82
83
84
85
86
87
88
89
90
def subscribe(
    self,
    event: WsEventType,
    handler: Handler,
    *,
    weak: bool = True,
) -> SubscriptionToken:
    """Register ``handler`` and return a token for this registration."""
    token = SubscriptionToken(event=event, id=uuid4())
    self._subscribers[event][token.id] = _Subscription.from_handler(token, handler, weak)
    return token
unsubscribe 
unsubscribe(token: SubscriptionToken) -> bool

Remove exactly the registration identified by token.

Source code in src/mcbe_ws_sdk/gateway/events.py 
92
93
94
def unsubscribe(self, token: SubscriptionToken) -> bool:
    """Remove exactly the registration identified by ``token``."""
    return self._subscribers[token.event].pop(token.id, None) is not None
handler_count 
handler_count(event: WsEventType) -> int

Return the number of live registrations for event.

Source code in src/mcbe_ws_sdk/gateway/events.py 
96
97
98
99
def handler_count(self, event: WsEventType) -> int:
    """Return the number of live registrations for ``event``."""
    self._prune_dead(event)
    return len(self._subscribers[event])
emit async 
emit(event: WsEventType, *args: Any, **kwargs: Any) -> None

Await live handlers in subscription order.

Each handler is isolated: an exception in one handler does not prevent subsequent handlers from being called.

Source code in src/mcbe_ws_sdk/gateway/events.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
async def emit(self, event: WsEventType, *args: Any, **kwargs: Any) -> None:
    """Await live handlers in subscription order.

    Each handler is isolated: an exception in one handler does not prevent
    subsequent handlers from being called.
    """
    subscribers = self._subscribers[event]
    for token_id in list(subscribers):
        subscription = subscribers.get(token_id)
        if subscription is None:
            continue
        handler = subscription.resolve()
        if handler is None:
            subscribers.pop(token_id, None)
            continue
        try:
            result = handler(*args, **kwargs)
            if inspect.isawaitable(result):
                await result
            elif result is not None:
                handler_name = getattr(handler, "__qualname__", repr(handler))
                raise TypeError(
                    f"Event handler {handler_name} for {event.value!r} must return None "
                    f"or an awaitable, got {type(result).__name__}"
                )
        except Exception:
            handler_name = getattr(handler, "__qualname__", repr(handler))
            structlog.get_logger(__name__).exception(
                "event_handler_failed",
                event_type=event.value,
                handler=handler_name,
            )

handler

MCBE WebSocket protocol handler.

Relocated from the main repo services/websocket/minecraft.py. Responsible for constructing subscribe messages, parsing inbound PlayerMessageEvent, resolving typed commands, and rendering the small set of on-screen status messages the connection lifecycle needs (info / success / error).

Deliberately host-config-agnostic: presentation strings (welcome template, status prefixes, colors) come in through a frozen :class:MessageSurfaceConfig value object instead of the host's MinecraftConfig. Command resolution is delegated to an injected :class:~mcbe_ws_sdk.command.registry.CommandRegistry.

create_chat_request was deliberately removed — it constructs the host's models.messages.ChatRequest and reads per-player provider/template state, which is now the host's concern (built inside its ConnectionHook / command dispatcher). The renderer methods like :meth:create_info_message remain because the host still needs to surface status lines; the host's HostSink delivers the resulting :class:TellrawMessage over the outbound delivery adapter.

Classes:

TellrawMessage dataclass 

TellrawMessage(text: str, color: str, target: str = '@a')

A structured outbound tellraw line.

Carries the plain text, the MC color, and the tellraw target so the delivery layer can chunk + serialize without re-parsing its own output.

MessageSurfaceConfig dataclass 

MessageSurfaceConfig(welcome_message_template: str = '-----------\n已连接到 MCBE WebSocket 网关\n连接 ID: {connection_id}...\n-----------', error_prefix: str = '❌ 错误: ', info_prefix: str = 'ℹ ', success_prefix: str = '✅ ', error_color: str = '§c', info_color: str = '§b', success_color: str = '§a')

Presentation strings the protocol handler renders status messages with.

A frozen value object so a host can supply any wording/colors via config without the handler importing the host's settings model.

MinecraftProtocolHandler 

MinecraftProtocolHandler(command_registry: CommandRegistry, surface: MessageSurfaceConfig | None = None)

Constructs / parses MCBE connection-lifecycle protocol messages.

__init__ takes the collaborators it needs (a command registry and a presentation surface) and never imports anything host-specific.

Methods:

Source code in src/mcbe_ws_sdk/gateway/handler.py 
85
86
87
88
89
90
91
def __init__(
    self,
    command_registry: CommandRegistry,
    surface: MessageSurfaceConfig | None = None,
) -> None:
    self.command_registry = command_registry
    self.surface = surface or MessageSurfaceConfig()
create_subscribe_message staticmethod 
create_subscribe_message() -> str

Build the subscribe payload string for PlayerMessage events.

Source code in src/mcbe_ws_sdk/gateway/handler.py 
93
94
95
96
97
@staticmethod
def create_subscribe_message() -> str:
    """Build the ``subscribe`` payload string for PlayerMessage events."""
    subscribe = MinecraftSubscribe.player_message()
    return subscribe.model_dump_json(exclude_none=True)
create_welcome_message 
create_welcome_message(*, connection_id: str) -> str

Render the post-connect welcome banner (plain text; wrap a message).

Source code in src/mcbe_ws_sdk/gateway/handler.py 
 99
100
101
102
103
104
105
106
107
def create_welcome_message(
    self,
    *,
    connection_id: str,
) -> str:
    """Render the post-connect welcome banner (plain text; wrap a message)."""
    return self.surface.welcome_message_template.format(
        connection_id=connection_id[:8],
    )
parse_player_message staticmethod 
parse_player_message(data: dict[str, Any]) -> PlayerMessageEvent | None

Parse an inbound PlayerMessage event out of a WS frame body.

Source code in src/mcbe_ws_sdk/gateway/handler.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
@staticmethod
def parse_player_message(data: dict[str, Any]) -> PlayerMessageEvent | None:
    """Parse an inbound ``PlayerMessage`` event out of a WS frame body."""
    try:
        header = data.get("header", {})
        event_name = header.get("eventName")
        if event_name != "PlayerMessage":
            return None
        body = data.get("body", {})
        return PlayerMessageEvent.from_event_body(body)
    except Exception:
        logger.warning(
            "parse_player_message_error",
            error_type="protocol_parse_failed",
        )
        return None
parse_command 
parse_command(message: str) -> tuple[str | None, str]

Resolve message to (command_type, content) via the registry.

Source code in src/mcbe_ws_sdk/gateway/handler.py 
126
127
128
def parse_command(self, message: str) -> tuple[str | None, str]:
    """Resolve ``message`` to ``(command_type, content)`` via the registry."""
    return self.command_registry.resolve(message)
parse_typed_command 
parse_typed_command(message: str) -> ParsedCommand | None

Resolve message to a typed :class:ParsedCommand, or None.

Source code in src/mcbe_ws_sdk/gateway/handler.py 
130
131
132
def parse_typed_command(self, message: str) -> ParsedCommand | None:
    """Resolve ``message`` to a typed :class:`ParsedCommand`, or ``None``."""
    return self.command_registry.resolve_parsed(message)
get_help_text 
get_help_text() -> str

Render the in-game help listing from the command registry.

Source code in src/mcbe_ws_sdk/gateway/handler.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
def get_help_text(self) -> str:
    """Render the in-game help listing from the command registry."""
    lines = ["可用命令:"]
    for prefix, _cmd_type, _aliases in self.command_registry.list_all_commands():
        cmd_config = self.command_registry.get_command_config(prefix)
        if cmd_config is None:
            continue
        desc = cmd_config.description
        usage = cmd_config.usage
        if usage:
            lines.append(f"• {prefix} {usage} - {desc}")
        else:
            lines.append(f"• {prefix} - {desc}")
    return "\n".join(lines)

hook

Connection lifecycle hook protocol.

The host (the main repo's McbeHost) implements these six points to inject the application-specific behaviour the gateway deliberately does NOT implement: prompt/context management, LLM dispatch, command routing, and addon linkage. NoOpHook is the gateway's built-in default — it defines the complete contract so a host can subclass and override only what it needs.

All hooks return None and are purely side-effecting. on_player_message receives an optional pre-parsed :class:~mcbe_ws_sdk.command.registry.ParsedCommand so the host does not need to re-run the registry.

Classes:

  • ConnectionHook –

    Lifecycle + protocol hooks the host injects into the gateway.

  • NoOpHook –

    Gateway default ConnectionHook — every hook is a no-op.

ConnectionHook

Bases: Protocol

Lifecycle + protocol hooks the host injects into the gateway.

Methods:

on_connected async 
on_connected(state: ConnectionState) -> None

Fired after the transport connection is established.

Source code in src/mcbe_ws_sdk/gateway/hook.py 
31
32
33
async def on_connected(self, state: ConnectionState) -> None:
    """Fired after the transport connection is established."""
    ...
on_disconnected async 
on_disconnected(state: ConnectionState) -> None

Fired on transport disconnect — host clears per-connection state here.

Source code in src/mcbe_ws_sdk/gateway/hook.py 
35
36
37
async def on_disconnected(self, state: ConnectionState) -> None:
    """Fired on transport disconnect — host clears per-connection state here."""
    ...
on_player_message async 
on_player_message(state: ConnectionState, player_event: PlayerMessageEvent, parsed: ParsedCommand | None = None) -> None

Inbound chat/scriptevent from a player.

parsed is the registry match for player_event.message when one exists; None means free-form chat (or no matching command prefix).

Source code in src/mcbe_ws_sdk/gateway/hook.py 
39
40
41
42
43
44
45
46
47
48
49
50
async def on_player_message(
    self,
    state: ConnectionState,
    player_event: PlayerMessageEvent,
    parsed: ParsedCommand | None = None,
) -> None:
    """Inbound chat/scriptevent from a player.

    ``parsed`` is the registry match for ``player_event.message`` when one
    exists; ``None`` means free-form chat (or no matching command prefix).
    """
    ...
on_ui_chat_reassembled async 
on_ui_chat_reassembled(state: ConnectionState, player_name: str, message: str) -> None

Fired when a fragmented UI_CHAT message is fully reassembled.

Source code in src/mcbe_ws_sdk/gateway/hook.py 
52
53
54
55
56
57
58
59
async def on_ui_chat_reassembled(
    self,
    state: ConnectionState,
    player_name: str,
    message: str,
) -> None:
    """Fired when a fragmented UI_CHAT message is fully reassembled."""
    ...
on_command_response async 
on_command_response(state: ConnectionState, response: MinecraftCommandResponse) -> None

Inbound commandResponse (resolves run_command futures).

Source code in src/mcbe_ws_sdk/gateway/hook.py 
61
62
63
64
65
66
67
async def on_command_response(
    self,
    state: ConnectionState,
    response: MinecraftCommandResponse,
) -> None:
    """Inbound ``commandResponse`` (resolves run_command futures)."""
    ...
on_error async 
on_error(state: ConnectionState, error: MinecraftErrorFrame) -> None

Inbound typed error frame.

Source code in src/mcbe_ws_sdk/gateway/hook.py 
69
70
71
async def on_error(self, state: ConnectionState, error: MinecraftErrorFrame) -> None:
    """Inbound typed ``error`` frame."""
    ...

NoOpHook

Gateway default ConnectionHook — every hook is a no-op.

messages

WebSocket gateway message value objects.

These are the display / delivery contract types the gateway produces on its event bus and routes through :class:~mcbe_ws_sdk.gateway.sink.ResponseSink. They are intentionally lean: they carry only the fields a gateway renderer or delivery path needs (what to say, to whom, how). They do NOT carry the agent's conversation framing (connection_id / BaseMessage.id / timestamp) — that host-specific metadata stays on the :class:ConnectionState the event is paired with. Keeping these types dependency-free is what lets the gateway layer type a sink without importing the agent's models.messages / core.queue.

Classes:

  • OutboundText –

    A text payload destined for a player (or all players) in the game.

  • SystemNotification –

    A host/system status line surfaced to a player (or all players).

OutboundText dataclass 

OutboundText(content: str, channel: str = 'content', sequence: int = 0, delivery: DeliveryMode = 'tellraw', player_name: str | None = None, target: str | None = None, message_id: str = 'server:data', id: UUID = uuid4())

A text payload destined for a player (or all players) in the game.

SystemNotification dataclass 

SystemNotification(level: NotificationLevel, message: str, player_name: str | None = None, id: UUID = uuid4())

A host/system status line surfaced to a player (or all players).

server_facade

Gateway entry point: the facade the host drives the SDK stack through.

:class:McbeServerFacade OWNS the WebSocket transport and the connection / protocol machinery (the :class:~mcbe_ws_sdk.gateway.connection.ConnectionManager response-sender loops, the :class:~mcbe_ws_sdk.gateway.handler.MinecraftProtocolHandler command parser, and the per-connection :class:~mcbe_ws_sdk.addon.service.AddonBridgeService sessions) but deliberately does NOT own any host application concern: there is no MessageBroker, no LLM worker, no provider selection. All of that is injected by the host via the :class:~mcbe_ws_sdk.gateway.hook.ConnectionHook lifecycle hooks and the :class:~mcbe_ws_sdk.gateway.sink.ResponseSink outbound routes — the same dependency-inversion the gateway already applies internally (see :class:NoOpHook, :class:DefaultResponseSink).

It mirrors the current main-repo WebSocketServer orchestration (services/websocket/server.py) — accept → handshake → subscribe → welcome → message loop → disconnect → shutdown — but inverted: the server receives its collaborators; it does not build the host-only ones.

Lifetime::

facade = McbeServerFacade(hook=my_hook, sink=my_sink)
await facade.run_lifetime()            # blocks until ``stop()`` / cancelled
# ... or drive it on an explicit host asyncio.Task and cancel it.

See docs/batch-d-scope.md section B for the authoritative spec.

Classes:

  • McbeServerFacade –

    Owns the WS transport + connection/protocol machinery; host injects the rest.

McbeServerFacade 

McbeServerFacade(*, settings: GatewaySettings | None = None, hook: ConnectionHook | None = None, sink: ResponseSink | None = None, addon: AddonBridgeService | None = None, registry: CommandRegistry | None = None, surface: MessageSurfaceConfig | None = None)

Owns the WS transport + connection/protocol machinery; host injects the rest.

The constructor takes ONLY keyword arguments and never builds a broker. Each None collapses to a gateway default so a host can stand up a working facade with McbeServerFacade() and override collaborators one at a time.

Methods:

  • run_lifetime –

    Bind a WebSocket server and serve until :meth:stop / task cancellation.

  • stop –

    Signal :meth:run_lifetime to unwind (idempotent).

Attributes:

Source code in src/mcbe_ws_sdk/gateway/server_facade.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def __init__(
    self,
    *,
    settings: GatewaySettings | None = None,
    hook: ConnectionHook | None = None,
    sink: ResponseSink | None = None,
    addon: AddonBridgeService | None = None,
    registry: CommandRegistry | None = None,
    surface: MessageSurfaceConfig | None = None,
) -> None:
    self._settings = settings if settings is not None else GatewaySettings()
    self._hook = hook if hook is not None else NoOpHook()
    self._sink = sink if sink is not None else DefaultResponseSink()
    self._registry = registry if registry is not None else CommandRegistry()

    self._handler = MinecraftProtocolHandler(self._registry, surface=surface)
    self._addon = addon if addon is not None else AddonBridgeService(self._settings.addon)
    self._addon.set_ui_chat_callback(self._on_ui_chat_reassembled)

    self._manager = ConnectionManager(
        sink=self._sink,
        event_bus=EventBus(),
        response_queue_maxsize=self._settings.websocket.response_queue_maxsize,
    )

    self._stopped = asyncio.Event()
    self._lifetime_started = False
    self._server: Any = None
manager property 
manager: ConnectionManager

The facade's owned connection manager.

handler property 
handler: MinecraftProtocolHandler

The facade's owned protocol handler (command parser + renderer).

addon property 
addon: AddonBridgeService

The facade's owned addon-bridge service instance.

run_lifetime async 
run_lifetime(host: str | None = None, port: int | None = None) -> None

Bind a WebSocket server and serve until :meth:stop / task cancellation.

host/port default to settings.websocket.host / port when None. Uses the websockets async-context-manager API (async with serve(...) as server), which works on both >=12 and the v13+ serve shape. Blocks on an internal asyncio.Event so an explicit stop() returns cleanly; the outer task being cancelled also unwinds into :meth:_graceful_shutdown.

Source code in src/mcbe_ws_sdk/gateway/server_facade.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
async def run_lifetime(
    self,
    host: str | None = None,
    port: int | None = None,
) -> None:
    """Bind a WebSocket server and serve until :meth:`stop` / task cancellation.

    ``host``/``port`` default to ``settings.websocket.host`` / ``port`` when
    ``None``. Uses the ``websockets`` async-context-manager API
    (``async with serve(...) as server``), which works on both ``>=12`` and
    the v13+ ``serve`` shape. Blocks on an internal ``asyncio.Event`` so an
    explicit ``stop()`` returns cleanly; the outer task being cancelled also
    unwinds into :meth:`_graceful_shutdown`.
    """
    if self._lifetime_started:
        raise FacadeLifecycleError("McbeServerFacade is single-use")
    self._lifetime_started = True
    serve_host = host if host is not None else self._settings.websocket.host
    serve_port = port if port is not None else self._settings.websocket.port
    transport = self._settings.websocket

    try:
        async with websockets.serve(
            self._on_connection,
            serve_host,
            serve_port,
            ping_interval=transport.ping_interval,
            ping_timeout=transport.ping_timeout,
            close_timeout=transport.close_timeout,
            max_size=transport.max_size,
            max_queue=transport.max_queue,
        ) as server:
            self._server = server
            logger.info(
                "facade_listening",
                host=serve_host,
                port=serve_port,
            )
            await self._stopped.wait()
    finally:
        try:
            await self._graceful_shutdown()
        finally:
            self._server = None
stop async 
stop() -> None

Signal :meth:run_lifetime to unwind (idempotent).

Source code in src/mcbe_ws_sdk/gateway/server_facade.py 
162
163
164
async def stop(self) -> None:
    """Signal :meth:`run_lifetime` to unwind (idempotent)."""
    self._stopped.set()

sink

Response routing sink + RouteEnvelope value object + DefaultResponseSink.

The response sender coroutine never builds Minecraft commands itself. Instead it asks a :class:ResponseSink to deliver a :class:RouteEnvelope, pushing the application-specific mapping entirely onto the host.

:class:DefaultResponseSink is the gateway's built-in base — it logs OUTBOUND_TEXT and SYSTEM_NOTIFICATION messages (metadata only, no player text) and does nothing else. A host wires an :class:~mcbe_ws_sdk.delivery.outbound.McbeOutboundDelivery here.

Classes:

  • ResponseKind –

    Message categories the response sender can route.

  • RouteEnvelope –

    A response message the response sender routes to a sink method.

  • ResponseSink –

    The two outbound delivery routes the response sender dispatches.

  • DefaultResponseSink –

    Gateway default sink: logs metadata, delivers nothing to game.

ResponseKind

Bases: Enum

Message categories the response sender can route.

RouteEnvelope dataclass 

RouteEnvelope(kind: ResponseKind, payload: OutboundText | SystemNotification)

A response message the response sender routes to a sink method.

Methods:

  • from_message –

    Classify a host response into a :class:RouteEnvelope.

from_message classmethod 
from_message(message: object) -> RouteEnvelope

Classify a host response into a :class:RouteEnvelope.

Accepts only the gateway's own value objects (OutboundText, SystemNotification) by type. Anything else is rejected — the response loop should never silently drop an unroutable message.

Source code in src/mcbe_ws_sdk/gateway/sink.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
@classmethod
def from_message(cls, message: object) -> RouteEnvelope:
    """Classify a host response into a :class:`RouteEnvelope`.

    Accepts only the gateway's own value objects (OutboundText,
    SystemNotification) by type. Anything else is rejected — the response
    loop should never silently drop an unroutable message.
    """
    if isinstance(message, OutboundText):
        return cls(ResponseKind.OUTBOUND_TEXT, message)
    if isinstance(message, SystemNotification):
        return cls(ResponseKind.SYSTEM_NOTIFICATION, message)
    raise TypeError(f"Unroutable response message: {type(message).__name__}")

ResponseSink

Bases: Protocol

The two outbound delivery routes the response sender dispatches.

dispatch is intentionally not part of this protocol: the connection manager routes by envelope kind and calls the matching on_* method directly so a duck-typed host only needs these two hooks.

DefaultResponseSink

Gateway default sink: logs metadata, delivers nothing to game.

on_outbound_text and on_system_notification log metadata (no player text, no command lines). A real host wires a :class:~mcbe_ws_sdk.delivery.outbound.McbeOutboundDelivery here.

dispatch remains as a non-protocol convenience for hosts/tests that want envelope-based routing; the manager never requires it.

Methods:

  • dispatch –

    Convenience router; not part of :class:ResponseSink.

dispatch async 
dispatch(state: ConnectionState, envelope: RouteEnvelope) -> None

Convenience router; not part of :class:ResponseSink.

Source code in src/mcbe_ws_sdk/gateway/sink.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
async def dispatch(self, state: ConnectionState, envelope: RouteEnvelope) -> None:
    """Convenience router; not part of :class:`ResponseSink`."""
    if envelope.kind is ResponseKind.OUTBOUND_TEXT:
        if not isinstance(envelope.payload, OutboundText):
            raise TypeError(
                "Expected OutboundText for OUTBOUND_TEXT kind, "
                f"got {type(envelope.payload).__name__}"
            )
        await self.on_outbound_text(state, envelope.payload)
        return
    if not isinstance(envelope.payload, SystemNotification):
        raise TypeError(
            "Expected SystemNotification for SYSTEM_NOTIFICATION kind, "
            f"got {type(envelope.payload).__name__}"
        )
    await self.on_system_notification(state, envelope.payload)

Addon bridge

mcbe_ws_sdk.addon

Addon bridge capability for the MCBE WebSocket SDK.

Modules:

  • service –

    Addon bridge service.

  • session –

    Addon bridge session management.

Classes:

AddonBridgeClient

Bases: Protocol

Face an Agent uses to reach the addon bridge.

Methods:

  • request –

    Issue a capability request and return the reassembled payload.

request async 

request(capability: str, payload: dict[str, Any]) -> dict[str, Any]

Issue a capability request and return the reassembled payload.

Source code in src/mcbe_ws_sdk/addon/service.py 
54
55
async def request(self, capability: str, payload: dict[str, Any]) -> dict[str, Any]:
    """Issue a capability request and return the reassembled payload."""

AddonBridgeService 

AddonBridgeService(settings: AddonBridgeSettings)

Manage the request/response lifecycle between Python and the addon.

Methods:

Source code in src/mcbe_ws_sdk/addon/service.py 
61
62
63
64
65
66
def __init__(self, settings: AddonBridgeSettings) -> None:
    self._settings = settings
    self._sessions: dict[UUID, AddonBridgeSession] = {}
    self._timeout_seconds = settings.timeout_seconds
    self._profile = settings.profile
    self._ui_chat_callback: UiChatCallback | None = None

create_client 

create_client(connection_id: UUID, send_command: CommandSender) -> AddonBridgeClient

Build a client bound to one connection.

Source code in src/mcbe_ws_sdk/addon/service.py 
68
69
70
71
72
73
74
def create_client(
    self,
    connection_id: UUID,
    send_command: CommandSender,
) -> AddonBridgeClient:
    """Build a client bound to one connection."""
    return ConnectionAddonBridgeClient(self, connection_id, send_command)

request_capability async 

request_capability(connection_id: UUID, capability: str, payload: dict[str, Any], send_command: CommandSender) -> dict[str, Any]

Send a bridge request and wait for the addon's chat callback.

Source code in src/mcbe_ws_sdk/addon/service.py 
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
async def request_capability(
    self,
    connection_id: UUID,
    capability: str,
    payload: dict[str, Any],
    send_command: CommandSender,
) -> dict[str, Any]:
    """Send a bridge request and wait for the addon's chat callback."""
    session = self._session_for(connection_id)
    request = session.create_request(capability=capability, payload=payload)
    command = encode_bridge_request(
        request_id=request.request_id,
        capability=capability,
        payload=payload,
        profile=self._profile,
    )
    payload_bytes = len(
        json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
    )
    # INFO stays free of payload/command content (may contain player data).
    logger.info(
        "bridge_request_outbound",
        request_id=request.request_id,
        capability=capability,
        payload_bytes=payload_bytes,
    )
    logger.debug(
        "bridge_request_outbound",
        connection_id=str(connection_id),
        request_id=request.request_id,
        capability=capability,
        payload=payload,
        command=command,
        timeout_seconds=self._timeout_seconds,
    )
    try:
        await send_command(command)
        try:
            result = await asyncio.wait_for(request.future, self._timeout_seconds)
        except TimeoutError as exc:
            logger.warning(
                "bridge_request_timeout",
                connection_id=str(connection_id),
                request_id=request.request_id,
                capability=capability,
                timeout_seconds=self._timeout_seconds,
            )
            raise BridgeTimeoutError(request.request_id) from exc
        logger.info(
            "bridge_request_resolved",
            connection_id=str(connection_id),
            request_id=request.request_id,
            capability=capability,
            result=result,
        )
        return result
    finally:
        session.cancel_request(request.request_id)

is_bridge_chat_message 

is_bridge_chat_message(sender: str, message: str) -> bool

True if the player chat message is an addon response fragment.

Source code in src/mcbe_ws_sdk/addon/service.py 
135
136
137
138
139
def is_bridge_chat_message(self, sender: str, message: str) -> bool:
    """True if the player chat message is an addon response fragment."""
    return sender == self._profile.bridge_sender and message.startswith(
        self._profile.bridge_response_prefix
    )

is_ui_chat_message 

is_ui_chat_message(sender: str, message: str) -> bool

True if the player chat message is a UI chat fragment.

Source code in src/mcbe_ws_sdk/addon/service.py 
141
142
143
144
145
def is_ui_chat_message(self, sender: str, message: str) -> bool:
    """True if the player chat message is a UI chat fragment."""
    return sender == self._profile.bridge_sender and message.startswith(
        self._profile.ui_chat_prefix
    )

handle_player_message async 

handle_player_message(connection_id: UUID, sender: str, message: str) -> AddonMessageResult

Route a routed message from the simulated player.

Source code in src/mcbe_ws_sdk/addon/service.py 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
async def handle_player_message(
    self, connection_id: UUID, sender: str, message: str
) -> AddonMessageResult:
    """Route a routed message from the simulated player."""
    if self.is_bridge_chat_message(sender, message):
        # Log every RESP chunk at INFO: these never reach the host hook because
        # the facade short-circuits bridge frames before on_player_message.
        logger.info(
            "bridge_chat_chunk_raw",
            connection_id=str(connection_id),
            sender=sender,
            message=message,
        )
        session = self._sessions.get(connection_id)
        if session is None:
            logger.warning(
                "bridge_chat_no_session",
                connection_id=str(connection_id),
            )
            return AddonMessageResult(handled=True)

        bridge_chunk = session.handle_chat_chunk(message)
        return AddonMessageResult(handled=True, bridge_chunk=bridge_chunk)

    if self.is_ui_chat_message(sender, message):
        logger.info(
            "ui_chat_chunk_raw",
            connection_id=str(connection_id),
            sender=sender,
            message=message,
        )
        session = self._session_for(connection_id)

        ui_chunk, ui_message = session.handle_ui_chat_chunk(message)
        if ui_message is not None and self._ui_chat_callback is not None:
            logger.info(
                "ui_chat_reassembled",
                connection_id=str(connection_id),
                player=ui_message.player_name,
                message_length=len(ui_message.message),
                callback_registered=True,
            )
            await self._ui_chat_callback(
                connection_id,
                ui_message.player_name,
                ui_message.message,
            )
        return AddonMessageResult(
            handled=True,
            ui_chunk=ui_chunk,
            ui_message=ui_message,
        )

    return AddonMessageResult(handled=False)

set_ui_chat_callback 

set_ui_chat_callback(callback: UiChatCallback) -> None

Register the UI chat message callback.

Source code in src/mcbe_ws_sdk/addon/service.py 
202
203
204
def set_ui_chat_callback(self, callback: UiChatCallback) -> None:
    """Register the UI chat message callback."""
    self._ui_chat_callback = callback

close_connection 

close_connection(connection_id: UUID) -> None

Tear down the per-connection session on disconnect.

Source code in src/mcbe_ws_sdk/addon/service.py 
206
207
208
209
210
def close_connection(self, connection_id: UUID) -> None:
    """Tear down the per-connection session on disconnect."""
    session = self._sessions.pop(connection_id, None)
    if session is not None:
        session.close()

ConnectionAddonBridgeClient 

ConnectionAddonBridgeClient(service: AddonBridgeService, connection_id: UUID, send_command: CommandSender)

Per-connection client bound to one bridge service instance.

Source code in src/mcbe_ws_sdk/addon/service.py 
223
224
225
226
227
228
229
230
231
def __init__(
    self,
    service: AddonBridgeService,
    connection_id: UUID,
    send_command: CommandSender,
) -> None:
    self._service = service
    self._connection_id = connection_id
    self._send_command = send_command

service

Addon bridge service.

Relocated from the main repo services/addon/service.py

Changes vs. the original:

  • No module-level _addon_bridge_service singleton and no get_addon_bridge_service() factory. The service is constructed with an explicit :class:AddonBridgeSettings and any number of independent instances may coexist.
  • Timeout and profile are taken from AddonBridgeSettings instead of a mutable global settings read.
  • is_bridge_chat_message / is_ui_chat_message use the configured profile.

Classes:

AddonBridgeClient

Bases: Protocol

Face an Agent uses to reach the addon bridge.

Methods:

  • request –

    Issue a capability request and return the reassembled payload.

request async 
request(capability: str, payload: dict[str, Any]) -> dict[str, Any]

Issue a capability request and return the reassembled payload.

Source code in src/mcbe_ws_sdk/addon/service.py 
54
55
async def request(self, capability: str, payload: dict[str, Any]) -> dict[str, Any]:
    """Issue a capability request and return the reassembled payload."""

AddonBridgeService 

AddonBridgeService(settings: AddonBridgeSettings)

Manage the request/response lifecycle between Python and the addon.

Methods:

Source code in src/mcbe_ws_sdk/addon/service.py 
61
62
63
64
65
66
def __init__(self, settings: AddonBridgeSettings) -> None:
    self._settings = settings
    self._sessions: dict[UUID, AddonBridgeSession] = {}
    self._timeout_seconds = settings.timeout_seconds
    self._profile = settings.profile
    self._ui_chat_callback: UiChatCallback | None = None
create_client 
create_client(connection_id: UUID, send_command: CommandSender) -> AddonBridgeClient

Build a client bound to one connection.

Source code in src/mcbe_ws_sdk/addon/service.py 
68
69
70
71
72
73
74
def create_client(
    self,
    connection_id: UUID,
    send_command: CommandSender,
) -> AddonBridgeClient:
    """Build a client bound to one connection."""
    return ConnectionAddonBridgeClient(self, connection_id, send_command)
request_capability async 
request_capability(connection_id: UUID, capability: str, payload: dict[str, Any], send_command: CommandSender) -> dict[str, Any]

Send a bridge request and wait for the addon's chat callback.

Source code in src/mcbe_ws_sdk/addon/service.py 
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
async def request_capability(
    self,
    connection_id: UUID,
    capability: str,
    payload: dict[str, Any],
    send_command: CommandSender,
) -> dict[str, Any]:
    """Send a bridge request and wait for the addon's chat callback."""
    session = self._session_for(connection_id)
    request = session.create_request(capability=capability, payload=payload)
    command = encode_bridge_request(
        request_id=request.request_id,
        capability=capability,
        payload=payload,
        profile=self._profile,
    )
    payload_bytes = len(
        json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
    )
    # INFO stays free of payload/command content (may contain player data).
    logger.info(
        "bridge_request_outbound",
        request_id=request.request_id,
        capability=capability,
        payload_bytes=payload_bytes,
    )
    logger.debug(
        "bridge_request_outbound",
        connection_id=str(connection_id),
        request_id=request.request_id,
        capability=capability,
        payload=payload,
        command=command,
        timeout_seconds=self._timeout_seconds,
    )
    try:
        await send_command(command)
        try:
            result = await asyncio.wait_for(request.future, self._timeout_seconds)
        except TimeoutError as exc:
            logger.warning(
                "bridge_request_timeout",
                connection_id=str(connection_id),
                request_id=request.request_id,
                capability=capability,
                timeout_seconds=self._timeout_seconds,
            )
            raise BridgeTimeoutError(request.request_id) from exc
        logger.info(
            "bridge_request_resolved",
            connection_id=str(connection_id),
            request_id=request.request_id,
            capability=capability,
            result=result,
        )
        return result
    finally:
        session.cancel_request(request.request_id)
is_bridge_chat_message 
is_bridge_chat_message(sender: str, message: str) -> bool

True if the player chat message is an addon response fragment.

Source code in src/mcbe_ws_sdk/addon/service.py 
135
136
137
138
139
def is_bridge_chat_message(self, sender: str, message: str) -> bool:
    """True if the player chat message is an addon response fragment."""
    return sender == self._profile.bridge_sender and message.startswith(
        self._profile.bridge_response_prefix
    )
is_ui_chat_message 
is_ui_chat_message(sender: str, message: str) -> bool

True if the player chat message is a UI chat fragment.

Source code in src/mcbe_ws_sdk/addon/service.py 
141
142
143
144
145
def is_ui_chat_message(self, sender: str, message: str) -> bool:
    """True if the player chat message is a UI chat fragment."""
    return sender == self._profile.bridge_sender and message.startswith(
        self._profile.ui_chat_prefix
    )
handle_player_message async 
handle_player_message(connection_id: UUID, sender: str, message: str) -> AddonMessageResult

Route a routed message from the simulated player.

Source code in src/mcbe_ws_sdk/addon/service.py 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
async def handle_player_message(
    self, connection_id: UUID, sender: str, message: str
) -> AddonMessageResult:
    """Route a routed message from the simulated player."""
    if self.is_bridge_chat_message(sender, message):
        # Log every RESP chunk at INFO: these never reach the host hook because
        # the facade short-circuits bridge frames before on_player_message.
        logger.info(
            "bridge_chat_chunk_raw",
            connection_id=str(connection_id),
            sender=sender,
            message=message,
        )
        session = self._sessions.get(connection_id)
        if session is None:
            logger.warning(
                "bridge_chat_no_session",
                connection_id=str(connection_id),
            )
            return AddonMessageResult(handled=True)

        bridge_chunk = session.handle_chat_chunk(message)
        return AddonMessageResult(handled=True, bridge_chunk=bridge_chunk)

    if self.is_ui_chat_message(sender, message):
        logger.info(
            "ui_chat_chunk_raw",
            connection_id=str(connection_id),
            sender=sender,
            message=message,
        )
        session = self._session_for(connection_id)

        ui_chunk, ui_message = session.handle_ui_chat_chunk(message)
        if ui_message is not None and self._ui_chat_callback is not None:
            logger.info(
                "ui_chat_reassembled",
                connection_id=str(connection_id),
                player=ui_message.player_name,
                message_length=len(ui_message.message),
                callback_registered=True,
            )
            await self._ui_chat_callback(
                connection_id,
                ui_message.player_name,
                ui_message.message,
            )
        return AddonMessageResult(
            handled=True,
            ui_chunk=ui_chunk,
            ui_message=ui_message,
        )

    return AddonMessageResult(handled=False)
set_ui_chat_callback 
set_ui_chat_callback(callback: UiChatCallback) -> None

Register the UI chat message callback.

Source code in src/mcbe_ws_sdk/addon/service.py 
202
203
204
def set_ui_chat_callback(self, callback: UiChatCallback) -> None:
    """Register the UI chat message callback."""
    self._ui_chat_callback = callback
close_connection 
close_connection(connection_id: UUID) -> None

Tear down the per-connection session on disconnect.

Source code in src/mcbe_ws_sdk/addon/service.py 
206
207
208
209
210
def close_connection(self, connection_id: UUID) -> None:
    """Tear down the per-connection session on disconnect."""
    session = self._sessions.pop(connection_id, None)
    if session is not None:
        session.close()

ConnectionAddonBridgeClient 

ConnectionAddonBridgeClient(service: AddonBridgeService, connection_id: UUID, send_command: CommandSender)

Per-connection client bound to one bridge service instance.

Source code in src/mcbe_ws_sdk/addon/service.py 
223
224
225
226
227
228
229
230
231
def __init__(
    self,
    service: AddonBridgeService,
    connection_id: UUID,
    send_command: CommandSender,
) -> None:
    self._service = service
    self._connection_id = connection_id
    self._send_command = send_command

session

Addon bridge session management.

Classes:

PendingAddonRequest dataclass 

PendingAddonRequest(request_id: str, capability: str, payload: dict[str, Any], future: Future[dict[str, Any]])

Pending state for a single bridge request.

AddonBridgeSession 

AddonBridgeSession(settings: AddonBridgeSettings, *, clock: Callable[[], float] = time.monotonic)

Maintain bridge requests and fragment buffers for one connection.

Methods:

  • create_request –

    Create and register a pending request.

  • handle_chat_chunk –

    Consume a chat fragment; resolve its request future once complete.

  • close –

    Close the session, failing every pending request.

  • handle_ui_chat_chunk –

    Consume a UI chat fragment; return the chunk and a completed message if available.

Source code in src/mcbe_ws_sdk/addon/session.py 
55
56
57
58
59
60
61
62
63
64
65
66
def __init__(
    self,
    settings: AddonBridgeSettings,
    *,
    clock: Callable[[], float] = time.monotonic,
) -> None:
    self._settings = settings
    self._profile = settings.profile
    self._clock = clock
    self._pending_requests: dict[str, PendingAddonRequest] = {}
    self._chunk_buffers: dict[str, ChunkBuffer[AddonBridgeChunk]] = {}
    self._ui_chat_chunk_buffers: dict[str, ChunkBuffer[UiChatChunk]] = {}
create_request 
create_request(capability: str, payload: dict[str, Any]) -> PendingAddonRequest

Create and register a pending request.

Source code in src/mcbe_ws_sdk/addon/session.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def create_request(
    self,
    capability: str,
    payload: dict[str, Any],
) -> PendingAddonRequest:
    """Create and register a pending request."""
    if len(self._pending_requests) >= self._settings.max_pending_requests:
        raise BridgeLimitError("maximum pending requests exceeded")
    loop = asyncio.get_running_loop()
    request = PendingAddonRequest(
        request_id=f"addon-{uuid4().hex}",
        capability=capability,
        payload=payload,
        future=loop.create_future(),
    )
    self._pending_requests[request.request_id] = request
    return request
handle_chat_chunk 
handle_chat_chunk(chunk_message: str) -> AddonBridgeChunk

Consume a chat fragment; resolve its request future once complete.

Source code in src/mcbe_ws_sdk/addon/session.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def handle_chat_chunk(self, chunk_message: str) -> AddonBridgeChunk:
    """Consume a chat fragment; resolve its request future once complete."""
    try:
        chunk = decode_bridge_chat_chunk(chunk_message, profile=self._profile)
    except ValueError as exc:
        protocol_error = ProtocolError(str(exc))
        request_id = self._pending_bridge_request_id(chunk_message)
        if request_id is not None and request_id in self._pending_requests:
            self._fail_bridge_request(request_id, protocol_error)
        raise protocol_error from exc
    if chunk.request_id not in self._pending_requests:
        return chunk

    try:
        buffer = self._accept_chunk(
            self._chunk_buffers,
            buffer_id=chunk.request_id,
            index=chunk.chunk_index,
            total=chunk.total_chunks,
            content=chunk.content,
            item=chunk,
        )
    except (BridgeLimitError, ProtocolError) as error:
        self._fail_bridge_request(chunk.request_id, error)
        raise
    if set(buffer.chunks) != set(range(1, buffer.total_chunks + 1)):
        return chunk

    complete = self._chunk_buffers.pop(chunk.request_id)
    try:
        response = reassemble_bridge_chunks(list(complete.chunks.values()))
    except ValueError as exc:
        protocol_error = ProtocolError(str(exc))
        self._fail_bridge_request(chunk.request_id, protocol_error)
        raise protocol_error from exc

    request = self._pending_requests.pop(chunk.request_id, None)
    if request is not None and not request.future.done():
        request.future.set_result(response.payload)
    return chunk
close 
close() -> None

Close the session, failing every pending request.

Source code in src/mcbe_ws_sdk/addon/session.py 
152
153
154
155
156
157
158
159
160
def close(self) -> None:
    """Close the session, failing every pending request."""
    pending_ids = list(self._pending_requests)
    for request_id in pending_ids:
        request = self._pending_requests.pop(request_id, None)
        self._chunk_buffers.pop(request_id, None)
        if request is not None and not request.future.done():
            request.future.set_exception(BridgeClosedError(request_id))
    self._ui_chat_chunk_buffers.clear()
handle_ui_chat_chunk 
handle_ui_chat_chunk(chunk_message: str) -> tuple[UiChatChunk, UiChatMessage | None]

Consume a UI chat fragment; return the chunk and a completed message if available.

Source code in src/mcbe_ws_sdk/addon/session.py 
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
def handle_ui_chat_chunk(self, chunk_message: str) -> tuple[UiChatChunk, UiChatMessage | None]:
    """Consume a UI chat fragment; return the chunk and a completed message if available."""
    try:
        chunk = decode_ui_chat_chunk(chunk_message, profile=self._profile)
    except ValueError as exc:
        raise ProtocolError(str(exc)) from exc

    buffer = self._accept_chunk(
        self._ui_chat_chunk_buffers,
        buffer_id=chunk.msg_id,
        index=chunk.chunk_index,
        total=chunk.total_chunks,
        content=chunk.content,
        item=chunk,
    )
    if set(buffer.chunks) != set(range(1, buffer.total_chunks + 1)):
        return chunk, None

    complete = self._ui_chat_chunk_buffers.pop(chunk.msg_id)
    try:
        ui_message = reassemble_ui_chat_chunks(list(complete.chunks.values()))
    except ValueError as exc:
        raise ProtocolError(str(exc)) from exc
    return chunk, ui_message

Flow control

mcbe_ws_sdk.flow

Flow control middleware for outbound MCBE command chunking.

Modules:

  • flow_control –

    Byte-safe outbound command chunking for MCBE commandLine limits.

Classes:

FlowControlMiddleware 

FlowControlMiddleware(settings: FlowControlSettings)

Unified flow control: split outbound long text into safe Minecraft commands.

Methods:

  • chunk_delay_for –

    Return inter-chunk delay in seconds for a chunking scenario.

  • chunk_tellraw –

    Split a long tellraw message into commandRequest JSON payload strings.

  • chunk_scriptevent –

    Split a long scriptevent payload into commandRequest JSON strings.

  • chunk_raw_command –

    Wrap a raw command as a one-element commandRequest JSON list.

Source code in src/mcbe_ws_sdk/flow/flow_control.py 
20
21
def __init__(self, settings: FlowControlSettings) -> None:
    self.settings = settings

chunk_delay_for 

chunk_delay_for(kind: str) -> float

Return inter-chunk delay in seconds for a chunking scenario.

kind is typically "tellraw" or "scriptevent". Unknown kinds yield 0.0 (no delay); the caller may treat that as an error.

Source code in src/mcbe_ws_sdk/flow/flow_control.py 
34
35
36
37
38
39
40
def chunk_delay_for(self, kind: str) -> float:
    """Return inter-chunk delay in seconds for a chunking scenario.

    ``kind`` is typically ``"tellraw"`` or ``"scriptevent"``. Unknown kinds
    yield ``0.0`` (no delay); the caller may treat that as an error.
    """
    return self.settings.chunk_delays.get(kind, 0.0)

chunk_tellraw 

chunk_tellraw(message: str, color: str = '§a', max_length: int | None = None, target: str = '@a') -> list[str]

Split a long tellraw message into commandRequest JSON payload strings.

Empty text returns []; the caller decides whether anything still must be sent. Each JSON commandLine text is at most max_length characters, and the final create_tellraw commandLine stays within the measured 461 B MCBE budget.

Source code in src/mcbe_ws_sdk/flow/flow_control.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
def chunk_tellraw(
    self,
    message: str,
    color: str = "§a",
    max_length: int | None = None,
    target: str = "@a",
) -> list[str]:
    """Split a long tellraw message into ``commandRequest`` JSON payload strings.

    Empty text returns ``[]``; the caller decides whether anything still must be
    sent. Each JSON ``commandLine`` text is at most ``max_length`` characters,
    and the final ``create_tellraw`` ``commandLine`` stays within the measured
    461 B MCBE budget.
    """
    if not message:
        return []

    max_len = self._get_max_length(max_length)
    text_parts = self._split_tellraw_text(
        message, color=color, target=target, max_length=max_len
    )

    payloads: list[str] = []
    for part in text_parts:
        command = MinecraftCommand.create_tellraw(part, color=color, target=target)
        payload = command.model_dump_json(exclude_none=True)
        self._assert_byte_safe(payload)
        payloads.append(payload)
    return payloads

chunk_scriptevent 

chunk_scriptevent(content: str, message_id: str = 'server:data', max_length: int | None = None) -> list[str]

Split a long scriptevent payload into commandRequest JSON strings.

Empty content returns []; the caller decides whether anything still must be sent. Each commandLine content segment is at most max_length characters, and total commandLine bytes stay within the measured 461 B MCBE budget.

Source code in src/mcbe_ws_sdk/flow/flow_control.py 
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def chunk_scriptevent(
    self,
    content: str,
    message_id: str = "server:data",
    max_length: int | None = None,
) -> list[str]:
    """Split a long scriptevent payload into ``commandRequest`` JSON strings.

    Empty content returns ``[]``; the caller decides whether anything still
    must be sent. Each ``commandLine`` content segment is at most
    ``max_length`` characters, and total ``commandLine`` bytes stay within
    the measured 461 B MCBE budget.
    """
    # Validate even empty content so invalid message_id is always rejected
    # before callers decide whether to send zero chunks.
    MinecraftCommand.create_scriptevent("", message_id)
    if not content:
        return []

    max_len = self._get_max_length(max_length)
    text_parts = self._split_by_command_fit(
        content,
        max_len,
        lambda part: MinecraftCommand.create_scriptevent(part, message_id).body.commandLine,
    )

    payloads: list[str] = []
    for part in text_parts:
        command = MinecraftCommand.create_scriptevent(part, message_id)
        payload = command.model_dump_json(exclude_none=True)
        self._assert_byte_safe(payload)
        payloads.append(payload)
    return payloads

chunk_raw_command 

chunk_raw_command(command: str) -> list[str]

Wrap a raw command as a one-element commandRequest JSON list.

Raw commands must not be truncated past the verb, or later fragments would be illegal. This method does not chunk: over-budget input raises FrameTooLargeError for the caller to handle.

Source code in src/mcbe_ws_sdk/flow/flow_control.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
def chunk_raw_command(self, command: str) -> list[str]:
    """Wrap a raw command as a one-element ``commandRequest`` JSON list.

    Raw commands must not be truncated past the verb, or later fragments would
    be illegal. This method **does not chunk**: over-budget input raises
    ``FrameTooLargeError`` for the caller to handle.
    """
    cmd = MinecraftCommand.create_raw(command)
    payload = cmd.model_dump_json(exclude_none=True)
    budget = self._byte_budget
    command_line_bytes = len(cmd.body.commandLine.encode("utf-8"))
    if command_line_bytes > budget:
        raise FrameTooLargeError(
            f"raw command too long in bytes "
            f"({command_line_bytes} > {budget}); "
            "cannot be safely chunked"
        )
    return [payload]

flow_control

Byte-safe outbound command chunking for MCBE commandLine limits.

Classes:

FlowControlMiddleware 

FlowControlMiddleware(settings: FlowControlSettings)

Unified flow control: split outbound long text into safe Minecraft commands.

Methods:

  • chunk_delay_for –

    Return inter-chunk delay in seconds for a chunking scenario.

  • chunk_tellraw –

    Split a long tellraw message into commandRequest JSON payload strings.

  • chunk_scriptevent –

    Split a long scriptevent payload into commandRequest JSON strings.

  • chunk_raw_command –

    Wrap a raw command as a one-element commandRequest JSON list.

Source code in src/mcbe_ws_sdk/flow/flow_control.py 
20
21
def __init__(self, settings: FlowControlSettings) -> None:
    self.settings = settings
chunk_delay_for 
chunk_delay_for(kind: str) -> float

Return inter-chunk delay in seconds for a chunking scenario.

kind is typically "tellraw" or "scriptevent". Unknown kinds yield 0.0 (no delay); the caller may treat that as an error.

Source code in src/mcbe_ws_sdk/flow/flow_control.py 
34
35
36
37
38
39
40
def chunk_delay_for(self, kind: str) -> float:
    """Return inter-chunk delay in seconds for a chunking scenario.

    ``kind`` is typically ``"tellraw"`` or ``"scriptevent"``. Unknown kinds
    yield ``0.0`` (no delay); the caller may treat that as an error.
    """
    return self.settings.chunk_delays.get(kind, 0.0)
chunk_tellraw 
chunk_tellraw(message: str, color: str = '§a', max_length: int | None = None, target: str = '@a') -> list[str]

Split a long tellraw message into commandRequest JSON payload strings.

Empty text returns []; the caller decides whether anything still must be sent. Each JSON commandLine text is at most max_length characters, and the final create_tellraw commandLine stays within the measured 461 B MCBE budget.

Source code in src/mcbe_ws_sdk/flow/flow_control.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
def chunk_tellraw(
    self,
    message: str,
    color: str = "§a",
    max_length: int | None = None,
    target: str = "@a",
) -> list[str]:
    """Split a long tellraw message into ``commandRequest`` JSON payload strings.

    Empty text returns ``[]``; the caller decides whether anything still must be
    sent. Each JSON ``commandLine`` text is at most ``max_length`` characters,
    and the final ``create_tellraw`` ``commandLine`` stays within the measured
    461 B MCBE budget.
    """
    if not message:
        return []

    max_len = self._get_max_length(max_length)
    text_parts = self._split_tellraw_text(
        message, color=color, target=target, max_length=max_len
    )

    payloads: list[str] = []
    for part in text_parts:
        command = MinecraftCommand.create_tellraw(part, color=color, target=target)
        payload = command.model_dump_json(exclude_none=True)
        self._assert_byte_safe(payload)
        payloads.append(payload)
    return payloads
chunk_scriptevent 
chunk_scriptevent(content: str, message_id: str = 'server:data', max_length: int | None = None) -> list[str]

Split a long scriptevent payload into commandRequest JSON strings.

Empty content returns []; the caller decides whether anything still must be sent. Each commandLine content segment is at most max_length characters, and total commandLine bytes stay within the measured 461 B MCBE budget.

Source code in src/mcbe_ws_sdk/flow/flow_control.py 
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def chunk_scriptevent(
    self,
    content: str,
    message_id: str = "server:data",
    max_length: int | None = None,
) -> list[str]:
    """Split a long scriptevent payload into ``commandRequest`` JSON strings.

    Empty content returns ``[]``; the caller decides whether anything still
    must be sent. Each ``commandLine`` content segment is at most
    ``max_length`` characters, and total ``commandLine`` bytes stay within
    the measured 461 B MCBE budget.
    """
    # Validate even empty content so invalid message_id is always rejected
    # before callers decide whether to send zero chunks.
    MinecraftCommand.create_scriptevent("", message_id)
    if not content:
        return []

    max_len = self._get_max_length(max_length)
    text_parts = self._split_by_command_fit(
        content,
        max_len,
        lambda part: MinecraftCommand.create_scriptevent(part, message_id).body.commandLine,
    )

    payloads: list[str] = []
    for part in text_parts:
        command = MinecraftCommand.create_scriptevent(part, message_id)
        payload = command.model_dump_json(exclude_none=True)
        self._assert_byte_safe(payload)
        payloads.append(payload)
    return payloads
chunk_raw_command 
chunk_raw_command(command: str) -> list[str]

Wrap a raw command as a one-element commandRequest JSON list.

Raw commands must not be truncated past the verb, or later fragments would be illegal. This method does not chunk: over-budget input raises FrameTooLargeError for the caller to handle.

Source code in src/mcbe_ws_sdk/flow/flow_control.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
def chunk_raw_command(self, command: str) -> list[str]:
    """Wrap a raw command as a one-element ``commandRequest`` JSON list.

    Raw commands must not be truncated past the verb, or later fragments would
    be illegal. This method **does not chunk**: over-budget input raises
    ``FrameTooLargeError`` for the caller to handle.
    """
    cmd = MinecraftCommand.create_raw(command)
    payload = cmd.model_dump_json(exclude_none=True)
    budget = self._byte_budget
    command_line_bytes = len(cmd.body.commandLine.encode("utf-8"))
    if command_line_bytes > budget:
        raise FrameTooLargeError(
            f"raw command too long in bytes "
            f"({command_line_bytes} > {budget}); "
            "cannot be safely chunked"
        )
    return [payload]

Protocol models

mcbe_ws_sdk.protocol

Minecraft protocol message models.

Modules:

  • minecraft –

    Minecraft protocol message models (WebSocket envelopes and helpers).

Classes:

PlayerMessageEvent

Bases: BaseModel

Parsed player chat/message event from the WebSocket stream.

Methods:

from_event_body classmethod 

from_event_body(body: dict[str, Any]) -> PlayerMessageEvent

Parse a PlayerMessageEvent from an event body dict.

Source code in src/mcbe_ws_sdk/protocol/minecraft.py 
195
196
197
198
199
200
201
202
203
@classmethod
def from_event_body(cls, body: dict[str, Any]) -> PlayerMessageEvent:
    """Parse a ``PlayerMessageEvent`` from an event body dict."""
    return cls(
        sender=body.get("sender", ""),
        message=body.get("message", ""),
        type=body.get("type"),
        receiver=body.get("receiver"),
    )

minecraft

Minecraft protocol message models (WebSocket envelopes and helpers).

Classes:

Functions:

MinecraftHeader

Bases: BaseModel

Minecraft WebSocket message header.

MinecraftOrigin

Bases: BaseModel

Command origin metadata on a request body.

MinecraftCommandBody

Bases: BaseModel

Body of a commandRequest frame.

MinecraftSubscribeBody

Bases: BaseModel

Body of a subscribe request.

MinecraftMessage

Bases: BaseModel

Generic Minecraft WebSocket message envelope.

MinecraftCommand

Bases: BaseModel

Minecraft commandRequest message.

Methods:

create_tellraw classmethod 
create_tellraw(message: str, color: str = '§a', target: str = '@a') -> MinecraftCommand

Build a tellraw command request.

Source code in src/mcbe_ws_sdk/protocol/minecraft.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
@classmethod
def create_tellraw(
    cls,
    message: str,
    color: str = "§a",
    target: str = "@a",
) -> MinecraftCommand:
    """Build a tellraw command request."""
    safe_message = sanitize_tellraw_text(message)
    rawtext = json.dumps(
        {"rawtext": [{"text": f"{color}{safe_message}"}]},
        ensure_ascii=False,
        separators=(",", ":"),
    )
    safe_target = sanitize_tellraw_target(target)
    command_line = f"tellraw {safe_target} {rawtext}"
    # Use the MinecraftCommandBody default origin (type="player").
    # "say" is a PlayerMessage event type, not a CommandRequest origin; using
    # it can leave remote clients without the tellraw in multiplayer.
    return cls(
        body=MinecraftCommandBody(
            commandLine=command_line,
        )
    )
create_scriptevent classmethod 
create_scriptevent(content: str, message_id: str = 'server:data') -> MinecraftCommand

Build a scriptevent command request.

Source code in src/mcbe_ws_sdk/protocol/minecraft.py 
150
151
152
153
154
155
156
157
158
@classmethod
def create_scriptevent(cls, content: str, message_id: str = "server:data") -> MinecraftCommand:
    """Build a scriptevent command request."""
    safe_message_id = validate_scriptevent_message_id(message_id)
    return cls(
        body=MinecraftCommandBody(
            commandLine=f"scriptevent {safe_message_id} {content}",
        )
    )
create_raw classmethod 
create_raw(command: str) -> MinecraftCommand

Build a raw command-line command request.

Source code in src/mcbe_ws_sdk/protocol/minecraft.py 
160
161
162
163
@classmethod
def create_raw(cls, command: str) -> MinecraftCommand:
    """Build a raw command-line command request."""
    return cls(body=MinecraftCommandBody(commandLine=command))

MinecraftSubscribe

Bases: BaseModel

Minecraft event subscription message.

Methods:

player_message classmethod 
player_message() -> MinecraftSubscribe

Subscribe to the PlayerMessage event.

Source code in src/mcbe_ws_sdk/protocol/minecraft.py 
179
180
181
182
@classmethod
def player_message(cls) -> MinecraftSubscribe:
    """Subscribe to the ``PlayerMessage`` event."""
    return cls(body=MinecraftSubscribeBody(eventName="PlayerMessage"))

PlayerMessageEvent

Bases: BaseModel

Parsed player chat/message event from the WebSocket stream.

Methods:

from_event_body classmethod 
from_event_body(body: dict[str, Any]) -> PlayerMessageEvent

Parse a PlayerMessageEvent from an event body dict.

Source code in src/mcbe_ws_sdk/protocol/minecraft.py 
195
196
197
198
199
200
201
202
203
@classmethod
def from_event_body(cls, body: dict[str, Any]) -> PlayerMessageEvent:
    """Parse a ``PlayerMessageEvent`` from an event body dict."""
    return cls(
        sender=body.get("sender", ""),
        message=body.get("message", ""),
        type=body.get("type"),
        receiver=body.get("receiver"),
    )

sanitize_tellraw_target 

sanitize_tellraw_target(target: str) -> str

Return a command-safe tellraw target without allowing command injection.

Player names are left unquoted whenever they contain no whitespace, quotes, or backslashes. This is required for non-ASCII Bedrock names (e.g. 玩家): wrapping them in double quotes makes the selector fail to match.

Source code in src/mcbe_ws_sdk/protocol/minecraft.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
def sanitize_tellraw_target(target: str) -> str:
    """Return a command-safe tellraw target without allowing command injection.

    Player names are left unquoted whenever they contain no whitespace, quotes,
    or backslashes. This is required for non-ASCII Bedrock names (e.g. ``玩家``):
    wrapping them in double quotes makes the selector fail to match.
    """
    target = target.strip() or "@a"
    if any(ord(char) < 0x20 or char == "\x7f" for char in target):
        raise ValueError("tellraw target contains control characters")
    if _TELLRAW_SELECTOR_RE.fullmatch(target):
        return target
    if not any(char in _TELLRAW_TARGET_QUOTE_CHARS for char in target):
        return target
    escaped = target.replace("\\", "\\\\").replace('"', '\\"')
    return f'"{escaped}"'

validate_scriptevent_message_id 

validate_scriptevent_message_id(message_id: str) -> str

Validate a scriptevent message id before embedding it in commandLine.

Source code in src/mcbe_ws_sdk/protocol/minecraft.py 
40
41
42
43
44
45
46
47
def validate_scriptevent_message_id(message_id: str) -> str:
    """Validate a scriptevent message id before embedding it in commandLine."""
    if not _SCRIPT_EVENT_MESSAGE_ID_RE.fullmatch(message_id):
        raise ValueError(
            "invalid scriptevent message_id; expected namespace:path matching "
            "^[a-z0-9_.-]+:[a-z0-9_./-]+$"
        )
    return message_id

sanitize_tellraw_text 

sanitize_tellraw_text(message: str) -> str

Escape tellraw text before embedding in the rawtext JSON value.

MCBE treats % as a format placeholder inside tellraw text; double it so literal percents render. Quote/backslash escaping is left to json.dumps.

Source code in src/mcbe_ws_sdk/protocol/minecraft.py 
54
55
56
57
58
59
60
def sanitize_tellraw_text(message: str) -> str:
    """Escape tellraw text before embedding in the rawtext JSON value.

    MCBE treats ``%`` as a format placeholder inside tellraw text; double it so
    literal percents render. Quote/backslash escaping is left to ``json.dumps``.
    """
    return message.replace("%", "%%")

Profiles

mcbe_ws_sdk.profiles

Protocol profiles for MCBE wire compatibility layers.

Modules:

Classes:

AddonBridgeProfile

Bases: Protocol

Wire-format contract for addon bridge interop.

mcbews_v1

mcbews v1 protocol profile.

Delivery

mcbe_ws_sdk.delivery

Outbound delivery adapter for MCBE WebSocket gateway.

Modules:

  • outbound –

    MCBE outbound delivery adapter.

Classes:

McbeOutboundDelivery 

McbeOutboundDelivery(*, connection_id: Any, send_payload: PayloadSender, settings: FlowControlSettings, log_raw_payloads: bool = False)

Chunk, throttle, send, and optionally log MCBE commandRequest payloads.

Methods:

  • send_payload –

    Send an already-built payload (subscribe, handshake, non-chunked paths).

  • send_tellraw –

    Send tellraw text and return the number of payloads actually transmitted.

  • send_scriptevent –

    Send scriptevent text and return the number of payloads actually transmitted.

  • send_raw_command –

    Send a raw command that must not be semantically split.

  • send_outbound_text –

    Route an OutboundText message through the delivery adapter.

  • send_system_notification –

    Route a SystemNotification through the delivery adapter.

  • send_chunked –

    Send payloads with inter-chunk delays.

Source code in src/mcbe_ws_sdk/delivery/outbound.py 
22
23
24
25
26
27
28
29
30
31
32
33
def __init__(
    self,
    *,
    connection_id: Any,
    send_payload: PayloadSender,
    settings: FlowControlSettings,
    log_raw_payloads: bool = False,
) -> None:
    self.connection_id = connection_id
    self._send_payload = send_payload
    self._flow = FlowControlMiddleware(settings)
    self._log_raw_payloads = log_raw_payloads

send_payload async 

send_payload(payload: str, source: str) -> None

Send an already-built payload (subscribe, handshake, non-chunked paths).

Source code in src/mcbe_ws_sdk/delivery/outbound.py 
39
40
41
async def send_payload(self, payload: str, source: str) -> None:
    """Send an already-built payload (subscribe, handshake, non-chunked paths)."""
    await self._send_one(payload, source)

send_tellraw async 

send_tellraw(message: str, color: str, source: str, target: str = '@a') -> int

Send tellraw text and return the number of payloads actually transmitted.

Source code in src/mcbe_ws_sdk/delivery/outbound.py 
43
44
45
46
47
48
49
50
51
52
53
async def send_tellraw(
    self,
    message: str,
    color: str,
    source: str,
    target: str = "@a",
) -> int:
    """Send tellraw text and return the number of payloads actually transmitted."""
    payloads = self.flow.chunk_tellraw(message, color=color, target=target)
    await self.send_chunked(payloads, "tellraw", source)
    return len(payloads)

send_scriptevent async 

send_scriptevent(content: str, message_id: str = 'server:data', source: str = 'scriptevent') -> int

Send scriptevent text and return the number of payloads actually transmitted.

Source code in src/mcbe_ws_sdk/delivery/outbound.py 
55
56
57
58
59
60
61
62
63
64
async def send_scriptevent(
    self,
    content: str,
    message_id: str = "server:data",
    source: str = "scriptevent",
) -> int:
    """Send scriptevent text and return the number of payloads actually transmitted."""
    payloads = self.flow.chunk_scriptevent(content, message_id)
    await self.send_chunked(payloads, "scriptevent", source)
    return len(payloads)

send_raw_command async 

send_raw_command(command: str, source: str = 'raw_command', before_send: Callable[[str], None] | None = None) -> str

Send a raw command that must not be semantically split.

Raises FrameTooLargeError when the command exceeds the byte budget.

Source code in src/mcbe_ws_sdk/delivery/outbound.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
async def send_raw_command(
    self,
    command: str,
    source: str = "raw_command",
    before_send: Callable[[str], None] | None = None,
) -> str:
    """Send a raw command that must not be semantically split.

    Raises ``FrameTooLargeError`` when the command exceeds the byte budget.
    """
    payload = self.flow.chunk_raw_command(command)[0]
    request_id = _request_id_from_payload(payload)
    if before_send is not None:
        before_send(request_id)
    await self._send_one(payload, source)
    return request_id

send_outbound_text async 

send_outbound_text(message: OutboundText) -> int

Route an OutboundText message through the delivery adapter.

Source code in src/mcbe_ws_sdk/delivery/outbound.py 
83
84
85
86
87
88
89
90
91
async def send_outbound_text(self, message: OutboundText) -> int:
    """Route an OutboundText message through the delivery adapter."""
    source = f"outbound_text:{message.channel}"
    if message.delivery == "scriptevent":
        return await self.send_scriptevent(
            message.content, message_id=message.message_id, source=source
        )
    target = message.target or message.player_name or "@a"
    return await self.send_tellraw(message.content, color="", source=source, target=target)

send_system_notification async 

send_system_notification(message: SystemNotification) -> int

Route a SystemNotification through the delivery adapter.

Source code in src/mcbe_ws_sdk/delivery/outbound.py 
 93
 94
 95
 96
 97
 98
 99
100
101
async def send_system_notification(self, message: SystemNotification) -> int:
    """Route a SystemNotification through the delivery adapter."""
    colors = {"info": "§b", "warning": "§e", "error": "§c"}
    return await self.send_tellraw(
        message.message,
        color=colors[message.level],
        source="system_notification",
        target=message.player_name or "@a",
    )

send_chunked async 

send_chunked(payloads: list[str], delay_kind: str, source: str, *, delay: float | None = None) -> None

Send payloads with inter-chunk delays.

When delay is provided it overrides flow.chunk_delay_for(delay_kind) so protocol profiles can inject their own cadence (e.g. profile.response_chunk_delay).

Source code in src/mcbe_ws_sdk/delivery/outbound.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
async def send_chunked(
    self,
    payloads: list[str],
    delay_kind: str,
    source: str,
    *,
    delay: float | None = None,
) -> None:
    """Send ``payloads`` with inter-chunk delays.

    When ``delay`` is provided it overrides ``flow.chunk_delay_for(delay_kind)``
    so protocol profiles can inject their own cadence (e.g.
    ``profile.response_chunk_delay``).
    """
    chunk_delay = self.flow.chunk_delay_for(delay_kind) if delay is None else delay
    for idx, payload in enumerate(payloads):
        await self._send_one(
            payload,
            source=f"{source}_chunked" if len(payloads) > 1 else source,
        )
        if len(payloads) > 1 and idx < len(payloads) - 1:
            await asyncio.sleep(chunk_delay)

outbound

MCBE outbound delivery adapter.

Classes:

McbeOutboundDelivery 

McbeOutboundDelivery(*, connection_id: Any, send_payload: PayloadSender, settings: FlowControlSettings, log_raw_payloads: bool = False)

Chunk, throttle, send, and optionally log MCBE commandRequest payloads.

Methods:

  • send_payload –

    Send an already-built payload (subscribe, handshake, non-chunked paths).

  • send_tellraw –

    Send tellraw text and return the number of payloads actually transmitted.

  • send_scriptevent –

    Send scriptevent text and return the number of payloads actually transmitted.

  • send_raw_command –

    Send a raw command that must not be semantically split.

  • send_outbound_text –

    Route an OutboundText message through the delivery adapter.

  • send_system_notification –

    Route a SystemNotification through the delivery adapter.

  • send_chunked –

    Send payloads with inter-chunk delays.

Source code in src/mcbe_ws_sdk/delivery/outbound.py 
22
23
24
25
26
27
28
29
30
31
32
33
def __init__(
    self,
    *,
    connection_id: Any,
    send_payload: PayloadSender,
    settings: FlowControlSettings,
    log_raw_payloads: bool = False,
) -> None:
    self.connection_id = connection_id
    self._send_payload = send_payload
    self._flow = FlowControlMiddleware(settings)
    self._log_raw_payloads = log_raw_payloads
send_payload async 
send_payload(payload: str, source: str) -> None

Send an already-built payload (subscribe, handshake, non-chunked paths).

Source code in src/mcbe_ws_sdk/delivery/outbound.py 
39
40
41
async def send_payload(self, payload: str, source: str) -> None:
    """Send an already-built payload (subscribe, handshake, non-chunked paths)."""
    await self._send_one(payload, source)
send_tellraw async 
send_tellraw(message: str, color: str, source: str, target: str = '@a') -> int

Send tellraw text and return the number of payloads actually transmitted.

Source code in src/mcbe_ws_sdk/delivery/outbound.py 
43
44
45
46
47
48
49
50
51
52
53
async def send_tellraw(
    self,
    message: str,
    color: str,
    source: str,
    target: str = "@a",
) -> int:
    """Send tellraw text and return the number of payloads actually transmitted."""
    payloads = self.flow.chunk_tellraw(message, color=color, target=target)
    await self.send_chunked(payloads, "tellraw", source)
    return len(payloads)
send_scriptevent async 
send_scriptevent(content: str, message_id: str = 'server:data', source: str = 'scriptevent') -> int

Send scriptevent text and return the number of payloads actually transmitted.

Source code in src/mcbe_ws_sdk/delivery/outbound.py 
55
56
57
58
59
60
61
62
63
64
async def send_scriptevent(
    self,
    content: str,
    message_id: str = "server:data",
    source: str = "scriptevent",
) -> int:
    """Send scriptevent text and return the number of payloads actually transmitted."""
    payloads = self.flow.chunk_scriptevent(content, message_id)
    await self.send_chunked(payloads, "scriptevent", source)
    return len(payloads)
send_raw_command async 
send_raw_command(command: str, source: str = 'raw_command', before_send: Callable[[str], None] | None = None) -> str

Send a raw command that must not be semantically split.

Raises FrameTooLargeError when the command exceeds the byte budget.

Source code in src/mcbe_ws_sdk/delivery/outbound.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
async def send_raw_command(
    self,
    command: str,
    source: str = "raw_command",
    before_send: Callable[[str], None] | None = None,
) -> str:
    """Send a raw command that must not be semantically split.

    Raises ``FrameTooLargeError`` when the command exceeds the byte budget.
    """
    payload = self.flow.chunk_raw_command(command)[0]
    request_id = _request_id_from_payload(payload)
    if before_send is not None:
        before_send(request_id)
    await self._send_one(payload, source)
    return request_id
send_outbound_text async 
send_outbound_text(message: OutboundText) -> int

Route an OutboundText message through the delivery adapter.

Source code in src/mcbe_ws_sdk/delivery/outbound.py 
83
84
85
86
87
88
89
90
91
async def send_outbound_text(self, message: OutboundText) -> int:
    """Route an OutboundText message through the delivery adapter."""
    source = f"outbound_text:{message.channel}"
    if message.delivery == "scriptevent":
        return await self.send_scriptevent(
            message.content, message_id=message.message_id, source=source
        )
    target = message.target or message.player_name or "@a"
    return await self.send_tellraw(message.content, color="", source=source, target=target)
send_system_notification async 
send_system_notification(message: SystemNotification) -> int

Route a SystemNotification through the delivery adapter.

Source code in src/mcbe_ws_sdk/delivery/outbound.py 
 93
 94
 95
 96
 97
 98
 99
100
101
async def send_system_notification(self, message: SystemNotification) -> int:
    """Route a SystemNotification through the delivery adapter."""
    colors = {"info": "§b", "warning": "§e", "error": "§c"}
    return await self.send_tellraw(
        message.message,
        color=colors[message.level],
        source="system_notification",
        target=message.player_name or "@a",
    )
send_chunked async 
send_chunked(payloads: list[str], delay_kind: str, source: str, *, delay: float | None = None) -> None

Send payloads with inter-chunk delays.

When delay is provided it overrides flow.chunk_delay_for(delay_kind) so protocol profiles can inject their own cadence (e.g. profile.response_chunk_delay).

Source code in src/mcbe_ws_sdk/delivery/outbound.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
async def send_chunked(
    self,
    payloads: list[str],
    delay_kind: str,
    source: str,
    *,
    delay: float | None = None,
) -> None:
    """Send ``payloads`` with inter-chunk delays.

    When ``delay`` is provided it overrides ``flow.chunk_delay_for(delay_kind)``
    so protocol profiles can inject their own cadence (e.g.
    ``profile.response_chunk_delay``).
    """
    chunk_delay = self.flow.chunk_delay_for(delay_kind) if delay is None else delay
    for idx, payload in enumerate(payloads):
        await self._send_one(
            payload,
            source=f"{source}_chunked" if len(payloads) > 1 else source,
        )
        if len(payloads) > 1 and idx < len(payloads) - 1:
            await asyncio.sleep(chunk_delay)

Command registry

mcbe_ws_sdk.command

Command parsing package: registry + parsed-command value object.

Modules:

  • registry –

    Command registry + parsed-command value object, relocated from the host app.

Classes:

CommandRegistry 

CommandRegistry(commands_config: Mapping[str, str | Mapping[str, Any]] | None = None)

Registry of command prefixes, types, and aliases.

Methods:

  • resolve –

    Resolve a message to (command_type, content).

  • resolve_parsed –

    Resolve a message to a typed command, including which token matched.

  • add_alias –

    Register a dynamic alias for an existing command prefix.

  • remove_alias –

    Remove a previously registered alias.

  • get_command_config –

    Return the loaded config for a primary command prefix.

  • get_aliases –

    Return all aliases for a command prefix.

  • list_all_commands –

    List all commands as (prefix, type, aliases) tuples.

  • get_command_prefix –

    Return the primary prefix for a command type, if registered.

Source code in src/mcbe_ws_sdk/command/registry.py 
43
44
45
46
47
48
49
50
def __init__(
    self,
    commands_config: Mapping[str, str | Mapping[str, Any]] | None = None,
) -> None:
    self._commands: dict[str, MinecraftCommandConfig] = {}
    self._alias_map: dict[str, str] = {}  # alias -> primary prefix
    self._type_to_prefix: dict[str, str] = {}  # command type -> primary prefix
    self._load_commands(commands_config or {})

resolve 

resolve(message: str) -> tuple[str | None, str]

Resolve a message to (command_type, content).

Source code in src/mcbe_ws_sdk/command/registry.py 
88
89
90
91
92
93
def resolve(self, message: str) -> tuple[str | None, str]:
    """Resolve a message to ``(command_type, content)``."""
    parsed = self.resolve_parsed(message)
    if parsed is None:
        return None, message
    return parsed.type, parsed.content

resolve_parsed 

resolve_parsed(message: str) -> ParsedCommand | None

Resolve a message to a typed command, including which token matched.

Source code in src/mcbe_ws_sdk/command/registry.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
def resolve_parsed(self, message: str) -> ParsedCommand | None:
    """Resolve a message to a typed command, including which token matched."""
    for prefix, cmd_config in self._commands.items():
        if self._matches_token(message, prefix):
            content = message[len(prefix) :].strip()
            return ParsedCommand(
                type=cmd_config.type,
                content=content,
                prefix=prefix,
                raw=message,
            )

    for alias, main_prefix in self._alias_map.items():
        if self._matches_token(message, alias):
            content = message[len(alias) :].strip()
            cmd_config = self._commands[main_prefix]
            return ParsedCommand(
                type=cmd_config.type,
                content=content,
                prefix=main_prefix,
                raw=message,
                matched_alias=alias,
            )

    return None

add_alias 

add_alias(command_prefix: str, alias: str) -> bool

Register a dynamic alias for an existing command prefix.

Source code in src/mcbe_ws_sdk/command/registry.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
def add_alias(self, command_prefix: str, alias: str) -> bool:
    """Register a dynamic alias for an existing command prefix."""
    if command_prefix not in self._commands:
        logger.warning("add_alias_command_not_found", prefix=command_prefix)
        return False

    if alias in self._alias_map:
        logger.warning("add_alias_already_exists", alias=alias)
        return False

    self._alias_map[alias] = command_prefix
    old = self._commands[command_prefix]
    self._commands[command_prefix] = replace(old, aliases=old.aliases + (alias,))

    logger.info("alias_added", prefix=command_prefix, alias=alias)
    return True

remove_alias 

remove_alias(alias: str) -> bool

Remove a previously registered alias.

Source code in src/mcbe_ws_sdk/command/registry.py 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
def remove_alias(self, alias: str) -> bool:
    """Remove a previously registered alias."""
    if alias not in self._alias_map:
        logger.warning("remove_alias_not_found", alias=alias)
        return False

    main_prefix = self._alias_map.pop(alias)
    old = self._commands[main_prefix]
    self._commands[main_prefix] = replace(
        old, aliases=tuple(a for a in old.aliases if a != alias)
    )

    logger.info("alias_removed", prefix=main_prefix, alias=alias)
    return True

get_command_config 

get_command_config(prefix: str) -> MinecraftCommandConfig | None

Return the loaded config for a primary command prefix.

Source code in src/mcbe_ws_sdk/command/registry.py 
161
162
163
def get_command_config(self, prefix: str) -> MinecraftCommandConfig | None:
    """Return the loaded config for a primary command prefix."""
    return self._commands.get(prefix)

get_aliases 

get_aliases(command_prefix: str) -> tuple[str, ...]

Return all aliases for a command prefix.

Source code in src/mcbe_ws_sdk/command/registry.py 
165
166
167
168
def get_aliases(self, command_prefix: str) -> tuple[str, ...]:
    """Return all aliases for a command prefix."""
    cmd = self._commands.get(command_prefix)
    return cmd.aliases if cmd else ()

list_all_commands 

list_all_commands() -> list[tuple[str, str, tuple[str, ...]]]

List all commands as (prefix, type, aliases) tuples.

Source code in src/mcbe_ws_sdk/command/registry.py 
170
171
172
def list_all_commands(self) -> list[tuple[str, str, tuple[str, ...]]]:
    """List all commands as ``(prefix, type, aliases)`` tuples."""
    return [(prefix, config.type, config.aliases) for prefix, config in self._commands.items()]

get_command_prefix 

get_command_prefix(cmd_type: str) -> str | None

Return the primary prefix for a command type, if registered.

Source code in src/mcbe_ws_sdk/command/registry.py 
174
175
176
def get_command_prefix(self, cmd_type: str) -> str | None:
    """Return the primary prefix for a command type, if registered."""
    return self._type_to_prefix.get(cmd_type)

registry

Command registry + parsed-command value object, relocated from the host app.

The registry turns a {prefix: type | config} mapping into a matcher that resolves an inbound message to a typed command. Matching is whole-word: a prefix/alias matches only when it is the entire message or is followed by whitespace, so #prefix_xyz does NOT match the #prefix token.

Classes:

MinecraftCommandConfig dataclass 

MinecraftCommandConfig(prefix: str, type: str, description: str, aliases: tuple[str, ...] = tuple(), usage: str | None = None)

A loaded command definition (prefix, type, aliases, description, usage).

CommandRegistry 

CommandRegistry(commands_config: Mapping[str, str | Mapping[str, Any]] | None = None)

Registry of command prefixes, types, and aliases.

Methods:

  • resolve –

    Resolve a message to (command_type, content).

  • resolve_parsed –

    Resolve a message to a typed command, including which token matched.

  • add_alias –

    Register a dynamic alias for an existing command prefix.

  • remove_alias –

    Remove a previously registered alias.

  • get_command_config –

    Return the loaded config for a primary command prefix.

  • get_aliases –

    Return all aliases for a command prefix.

  • list_all_commands –

    List all commands as (prefix, type, aliases) tuples.

  • get_command_prefix –

    Return the primary prefix for a command type, if registered.

Source code in src/mcbe_ws_sdk/command/registry.py 
43
44
45
46
47
48
49
50
def __init__(
    self,
    commands_config: Mapping[str, str | Mapping[str, Any]] | None = None,
) -> None:
    self._commands: dict[str, MinecraftCommandConfig] = {}
    self._alias_map: dict[str, str] = {}  # alias -> primary prefix
    self._type_to_prefix: dict[str, str] = {}  # command type -> primary prefix
    self._load_commands(commands_config or {})
resolve 
resolve(message: str) -> tuple[str | None, str]

Resolve a message to (command_type, content).

Source code in src/mcbe_ws_sdk/command/registry.py 
88
89
90
91
92
93
def resolve(self, message: str) -> tuple[str | None, str]:
    """Resolve a message to ``(command_type, content)``."""
    parsed = self.resolve_parsed(message)
    if parsed is None:
        return None, message
    return parsed.type, parsed.content
resolve_parsed 
resolve_parsed(message: str) -> ParsedCommand | None

Resolve a message to a typed command, including which token matched.

Source code in src/mcbe_ws_sdk/command/registry.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
def resolve_parsed(self, message: str) -> ParsedCommand | None:
    """Resolve a message to a typed command, including which token matched."""
    for prefix, cmd_config in self._commands.items():
        if self._matches_token(message, prefix):
            content = message[len(prefix) :].strip()
            return ParsedCommand(
                type=cmd_config.type,
                content=content,
                prefix=prefix,
                raw=message,
            )

    for alias, main_prefix in self._alias_map.items():
        if self._matches_token(message, alias):
            content = message[len(alias) :].strip()
            cmd_config = self._commands[main_prefix]
            return ParsedCommand(
                type=cmd_config.type,
                content=content,
                prefix=main_prefix,
                raw=message,
                matched_alias=alias,
            )

    return None
add_alias 
add_alias(command_prefix: str, alias: str) -> bool

Register a dynamic alias for an existing command prefix.

Source code in src/mcbe_ws_sdk/command/registry.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
def add_alias(self, command_prefix: str, alias: str) -> bool:
    """Register a dynamic alias for an existing command prefix."""
    if command_prefix not in self._commands:
        logger.warning("add_alias_command_not_found", prefix=command_prefix)
        return False

    if alias in self._alias_map:
        logger.warning("add_alias_already_exists", alias=alias)
        return False

    self._alias_map[alias] = command_prefix
    old = self._commands[command_prefix]
    self._commands[command_prefix] = replace(old, aliases=old.aliases + (alias,))

    logger.info("alias_added", prefix=command_prefix, alias=alias)
    return True
remove_alias 
remove_alias(alias: str) -> bool

Remove a previously registered alias.

Source code in src/mcbe_ws_sdk/command/registry.py 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
def remove_alias(self, alias: str) -> bool:
    """Remove a previously registered alias."""
    if alias not in self._alias_map:
        logger.warning("remove_alias_not_found", alias=alias)
        return False

    main_prefix = self._alias_map.pop(alias)
    old = self._commands[main_prefix]
    self._commands[main_prefix] = replace(
        old, aliases=tuple(a for a in old.aliases if a != alias)
    )

    logger.info("alias_removed", prefix=main_prefix, alias=alias)
    return True
get_command_config 
get_command_config(prefix: str) -> MinecraftCommandConfig | None

Return the loaded config for a primary command prefix.

Source code in src/mcbe_ws_sdk/command/registry.py 
161
162
163
def get_command_config(self, prefix: str) -> MinecraftCommandConfig | None:
    """Return the loaded config for a primary command prefix."""
    return self._commands.get(prefix)
get_aliases 
get_aliases(command_prefix: str) -> tuple[str, ...]

Return all aliases for a command prefix.

Source code in src/mcbe_ws_sdk/command/registry.py 
165
166
167
168
def get_aliases(self, command_prefix: str) -> tuple[str, ...]:
    """Return all aliases for a command prefix."""
    cmd = self._commands.get(command_prefix)
    return cmd.aliases if cmd else ()
list_all_commands 
list_all_commands() -> list[tuple[str, str, tuple[str, ...]]]

List all commands as (prefix, type, aliases) tuples.

Source code in src/mcbe_ws_sdk/command/registry.py 
170
171
172
def list_all_commands(self) -> list[tuple[str, str, tuple[str, ...]]]:
    """List all commands as ``(prefix, type, aliases)`` tuples."""
    return [(prefix, config.type, config.aliases) for prefix, config in self._commands.items()]
get_command_prefix 
get_command_prefix(cmd_type: str) -> str | None

Return the primary prefix for a command type, if registered.

Source code in src/mcbe_ws_sdk/command/registry.py 
174
175
176
def get_command_prefix(self, cmd_type: str) -> str | None:
    """Return the primary prefix for a command type, if registered."""
    return self._type_to_prefix.get(cmd_type)

Config & logging

mcbe_ws_sdk.config

Classes:

WebsocketTransportConfig dataclass 

WebsocketTransportConfig(host: str = '0.0.0.0', port: int = 8080, ping_interval: float | None = 30.0, ping_timeout: float | None = 15.0, close_timeout: float = 15.0, max_size: int | None = 10 * 1024 * 1024, max_queue: int | None = 32, response_queue_maxsize: int = 256)

Transport knobs for the facade's websockets.serve lifetime.

mcbe_ws_sdk.logging

Optional console logging helpers for hosts and examples.

The SDK itself only emits structured events via structlog.get_logger. Hosts that want a readable console should call :func:configure_logging once at process start. Defaults match the compact style used by the parent MCBE AI Agent host: no level/event column padding, local timestamps, key=value fields.

Functions:

  • configure_logging –

    Configure structlog + stdlib logging for compact console output.

configure_logging 

configure_logging(level: LogLevel | str = 'INFO', *, colors: bool | None = None) -> None

Configure structlog + stdlib logging for compact console output.

Parameters

level: Root log level name (default INFO). colors: Force colour on/off. None (default) enables colour only when stdout is a TTY.

Source code in src/mcbe_ws_sdk/logging.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
def configure_logging(
    level: LogLevel | str = "INFO",
    *,
    colors: bool | None = None,
) -> None:
    """Configure structlog + stdlib logging for compact console output.

    Parameters
    ----------
    level:
        Root log level name (default ``INFO``).
    colors:
        Force colour on/off. ``None`` (default) enables colour only when stdout
        is a TTY.
    """
    level_name = str(level).upper()
    log_level = getattr(logging, level_name, None)
    if not isinstance(log_level, int):
        raise ValueError(f"unknown log level: {level!r}")

    use_colors = sys.stdout.isatty() if colors is None else colors

    shared_processors: list[structlog.types.Processor] = [
        structlog.contextvars.merge_contextvars,
        structlog.stdlib.add_log_level,
        structlog.processors.TimeStamper(fmt="%Y-%m-%d %H:%M:%S", utc=False),
        structlog.processors.StackInfoRenderer(),
        structlog.processors.format_exc_info,
        structlog.processors.UnicodeDecoder(),
    ]

    structlog.configure(
        processors=[
            *shared_processors,
            structlog.stdlib.ProcessorFormatter.wrap_for_formatter,
        ],
        wrapper_class=structlog.stdlib.BoundLogger,
        context_class=dict,
        logger_factory=structlog.stdlib.LoggerFactory(),
        cache_logger_on_first_use=True,
    )

    renderer: structlog.types.Processor
    if use_colors:
        renderer = structlog.dev.ConsoleRenderer(
            colors=True,
            pad_level=False,
            pad_event_to=0,
        )
    else:
        # Keep the same key=value layout when piping / redirecting; hosts that
        # want JSON can wire their own renderer.
        renderer = structlog.dev.ConsoleRenderer(
            colors=False,
            pad_level=False,
            pad_event_to=0,
        )

    formatter = structlog.stdlib.ProcessorFormatter(
        processor=renderer,
        foreign_pre_chain=shared_processors,
    )

    handler = logging.StreamHandler(sys.stdout)
    handler.setLevel(log_level)
    handler.setFormatter(formatter)

    root = logging.getLogger()
    root.handlers.clear()
    root.setLevel(log_level)
    root.addHandler(handler)

Errors

mcbe_ws_sdk.errors

Public exception hierarchy for mcbe-ws-sdk.

Classes:

McbeWsSdkError

Bases: Exception

Base class for SDK errors.

ConfigurationError

Bases: McbeWsSdkError, ValueError

Raised when SDK settings are invalid.

ProtocolError

Bases: McbeWsSdkError, ValueError

Raised when a protocol contract is violated.

FrameTooLargeError

Bases: ProtocolError

Raised when a frame cannot fit within its configured byte budget.

BridgeError

Bases: McbeWsSdkError

Base class for add-on bridge errors.

BridgeTimeoutError 

BridgeTimeoutError(request_id: str)

Bases: BridgeError

Raised when an add-on bridge request times out.

Source code in src/mcbe_ws_sdk/errors.py 
27
28
29
def __init__(self, request_id: str) -> None:
    super().__init__(f"Bridge request timed out: {request_id}")
    self.request_id = request_id

BridgeClosedError 

BridgeClosedError(request_id: str)

Bases: BridgeError

Raised when an add-on bridge is closed.

Source code in src/mcbe_ws_sdk/errors.py 
35
36
37
def __init__(self, request_id: str) -> None:
    super().__init__(f"Bridge connection closed: {request_id}")
    self.request_id = request_id

BridgeLimitError

Bases: BridgeError, ProtocolError

Raised when an add-on bridge protocol limit is exceeded.

FacadeLifecycleError

Bases: McbeWsSdkError, RuntimeError

Raised for invalid server facade lifecycle transitions.