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 –

    MCBEWS/1 protocol profile exports.

  • protocol –

    Minecraft protocol message models.

Classes:

Functions:

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 
71
72
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 
78
79
80
81
82
83
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 
85
86
87
88
89
90
91
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 
 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
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
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
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,
        channel="capability",
        payload_bytes=payload_bytes,
    )
    logger.debug(
        "bridge_request_outbound",
        connection_id=str(connection_id),
        request_id=request.request_id,
        capability=capability,
        command_bytes=len(command.encode("utf-8")),
        timeout_seconds=self._timeout_seconds,
    )
    try:
        try:
            await send_command(command)
        except Exception as exc:
            # Fail-fast before waiting: host rejected the outbound frame
            # (e.g. FrameTooLarge) or the connection is gone. Waiting for a
            # RESP that can never arrive only produces a misleading timeout.
            logger.warning(
                "bridge_request_send_failed",
                connection_id=str(connection_id),
                request_id=request.request_id,
                channel="capability",
                payload_bytes=payload_bytes,
                error_type=type(exc).__name__,
            )
            raise
        logger.info(
            "bridge_request_sent",
            connection_id=str(connection_id),
            request_id=request.request_id,
            channel="capability",
            payload_bytes=payload_bytes,
        )
        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,
                channel="capability",
                payload_bytes=payload_bytes,
            )
            raise BridgeTimeoutError(request.request_id) from exc
        logger.info(
            "bridge_request_resolved",
            connection_id=str(connection_id),
            request_id=request.request_id,
            channel="capability",
            result_type=type(result).__name__,
        )
        logger.debug(
            "bridge_request_resolved_payload",
            connection_id=str(connection_id),
            request_id=request.request_id,
            capability=capability,
            result_type=type(result).__name__,
        )
        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 
179
180
181
182
183
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.trusted_bridge_player_name and message.startswith(
        f"{self._profile.capability_response_chat_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 
185
186
187
188
189
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.trusted_bridge_player_name and message.startswith(
        f"{self._profile.ui_chat_chunk_prefix}|"
    )

is_tool_player_channel_message 

is_tool_player_channel_message(message: str) -> bool

Return whether message is reserved for a ToolPlayer channel.

Source code in src/mcbe_ws_sdk/addon/service.py 
191
192
193
194
def is_tool_player_channel_message(self, message: str) -> bool:
    """Return whether *message* is reserved for a ToolPlayer channel."""

    return is_tool_player_channel_message(message, profile=self._profile)

classify_tool_player_message 

classify_tool_player_message(sender: str, message: str) -> ToolPlayerMessage | None

Authenticate and classify a reserved ToolPlayer message.

Source code in src/mcbe_ws_sdk/addon/service.py 
196
197
198
199
def classify_tool_player_message(self, sender: str, message: str) -> ToolPlayerMessage | None:
    """Authenticate and classify a reserved ToolPlayer message."""

    return classify_tool_player_message(sender, message, profile=self._profile)

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 
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
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):
        # Keep INFO body-free: these frames 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),
            channel="capability",
            message_bytes=len(message.encode("utf-8")),
        )
        session = self._sessions.get(connection_id)
        if session is None:
            logger.warning(
                "bridge_chat_no_session",
                connection_id=str(connection_id),
                channel="capability",
            )
            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),
            channel="ui_chat",
            message_bytes=len(message.encode("utf-8")),
        )
        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),
                channel="ui_chat",
                message_bytes=len(ui_message.message.encode("utf-8")),
            )
            await self._invoke_ui_chat_callback(connection_id, ui_message)
        return AddonMessageResult(
            handled=True,
            ui_chunk=ui_chunk,
            ui_message=ui_message,
        )

    if self.is_tool_player_channel_message(message):
        # Session and approval frames are decoded only after the trusted
        # sender gate.  The facade uses this branch to deliver typed control
        # messages to the Host hook; they never fall through as player chat.
        control_message = self.classify_tool_player_message(sender, message)
        if control_message is None:
            return AddonMessageResult(handled=True)
        return AddonMessageResult(handled=True, control_message=control_message)

    return AddonMessageResult(handled=False)

set_ui_chat_callback 

set_ui_chat_callback(callback: UiChatCallback) -> None

Register the typed UI chat message callback.

Source code in src/mcbe_ws_sdk/addon/service.py 
261
262
263
264
265
266
def set_ui_chat_callback(
    self,
    callback: UiChatCallback,
) -> None:
    """Register the typed UI chat message callback."""
    self._ui_chat_callback = callback

set_legacy_ui_chat_callback 

set_legacy_ui_chat_callback(callback: LegacyUiChatCallback) -> None

Register a legacy callback through an explicit migration adapter.

Source code in src/mcbe_ws_sdk/addon/service.py 
268
269
270
def set_legacy_ui_chat_callback(self, callback: LegacyUiChatCallback) -> None:
    """Register a legacy callback through an explicit migration adapter."""
    self._ui_chat_callback = LegacyUiChatCallbackAdapter(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 
279
280
281
282
283
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 
296
297
298
299
300
301
302
303
304
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

LegacyUiChatCallbackAdapter 

LegacyUiChatCallbackAdapter(callback: LegacyUiChatCallback)

Adapt the pre-0.2.0 three-argument callback to the typed DTO callback.

Source code in src/mcbe_ws_sdk/addon/service.py 
52
53
def __init__(self, callback: LegacyUiChatCallback) -> None:
    self._callback = callback

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]

AddonControlHook

Bases: Protocol

Optional hook for authenticated session/approval ToolPlayer DTOs.

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 
33
34
35
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 
37
38
39
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 
41
42
43
44
45
46
47
48
49
50
51
52
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, message: UiChatMessage) -> None

Fired when a fragmented UI_CHAT message is fully reassembled.

Source code in src/mcbe_ws_sdk/gateway/hook.py 
54
55
56
57
async def on_ui_chat_reassembled(self, state: ConnectionState, message: UiChatMessage) -> 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 
59
60
61
62
63
64
65
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 
67
68
69
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,
            )

LegacyUiChatHook

Bases: Protocol

Pre-0.2.0 UI callback shape used only through an explicit adapter.

LegacyUiChatHookAdapter 

LegacyUiChatHookAdapter(legacy_hook: LegacyUiChatHook)

Adapt a legacy three-argument UI hook to the typed callback boundary.

Source code in src/mcbe_ws_sdk/gateway/hook.py 
125
126
def __init__(self, legacy_hook: LegacyUiChatHook) -> None:
    self._legacy_hook = legacy_hook

McbeServerFacade 

McbeServerFacade(*, settings: GatewaySettings | None = None, hook: ConnectionHook | None = None, legacy_ui_chat_hook: LegacyUiChatHook | 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 
 72
 73
 74
 75
 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
def __init__(
    self,
    *,
    settings: GatewaySettings | None = None,
    hook: ConnectionHook | None = None,
    legacy_ui_chat_hook: LegacyUiChatHook | 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._legacy_ui_chat_hook = (
        LegacyUiChatHookAdapter(legacy_ui_chat_hook)
        if legacy_ui_chat_hook is not None
        else None
    )
    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 
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
161
162
163
164
165
166
167
168
169
170
171
172
173
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 
175
176
177
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.

McbewsV1Profile dataclass 

McbewsV1Profile(*, capability_request_script_event_id: str | None = None, capability_response_chat_prefix: str | None = None, ui_chat_chunk_prefix: str | None = None, session_request_chat_prefix: str | None = None, session_request_script_event_id: str | None = None, session_response_script_event_id: str | None = None, text_response_script_event_id: str | None = None, approval_allow_chat_prefix: str | None = None, approval_deny_chat_prefix: str | None = None, trusted_bridge_player_name: str | None = None, capability_request_schema_version: int | None = None, session_schema_version: int | None = None, text_response_framing_version: int | None = None, ddui_persistence_version: int | None = None, command_line_byte_budget: int | None = None, upstream_max_content_code_points: int | None = None, response_max_buffers: int | None = None, response_max_chunks_per_message: int | None = None, response_max_message_bytes: int | None = None, response_max_total_buffer_bytes: int | None = None, response_buffer_ttl_ms: int | None = None, session_response_max_command_bytes: int | None = None, response_chunk_delay: float = 0.15, response_prelude_delay: float = 0.5, bridge_request_message_id: str | None = None, bridge_response_prefix: str | None = None, ui_chat_prefix: str | None = None, bridge_sender: str | None = None, response_message_id: str | None = None, session_request_message_id: str | None = None, session_response_message_id: str | None = None, request_version: int | None = None)

Concrete MCBEWS/1 wire profile.

The semantic field names are the protocol authority. Wire identifiers, schema versions and safety bounds come from the installed manifest and cannot be replaced by a runtime profile seam. The empirical command budget may be lowered for a deployment and response delays remain operational knobs. Historical names are accepted for one migration cycle only when they still select the manifest value.

Source code in src/mcbe_ws_sdk/profiles/mcbews_v1/profile.py 
 70
 71
 72
 73
 74
 75
 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
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
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
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
def __init__(
    self,
    *,
    capability_request_script_event_id: str | None = None,
    capability_response_chat_prefix: str | None = None,
    ui_chat_chunk_prefix: str | None = None,
    session_request_chat_prefix: str | None = None,
    session_request_script_event_id: str | None = None,
    session_response_script_event_id: str | None = None,
    text_response_script_event_id: str | None = None,
    approval_allow_chat_prefix: str | None = None,
    approval_deny_chat_prefix: str | None = None,
    trusted_bridge_player_name: str | None = None,
    capability_request_schema_version: int | None = None,
    session_schema_version: int | None = None,
    text_response_framing_version: int | None = None,
    ddui_persistence_version: int | None = None,
    command_line_byte_budget: int | None = None,
    upstream_max_content_code_points: int | None = None,
    response_max_buffers: int | None = None,
    response_max_chunks_per_message: int | None = None,
    response_max_message_bytes: int | None = None,
    response_max_total_buffer_bytes: int | None = None,
    response_buffer_ttl_ms: int | None = None,
    session_response_max_command_bytes: int | None = None,
    response_chunk_delay: float = 0.15,
    response_prelude_delay: float = 0.5,
    # Deprecated aliases.  They are deliberately not stored as independent
    # values, so alias and semantic field parity cannot drift.
    bridge_request_message_id: str | None = None,
    bridge_response_prefix: str | None = None,
    ui_chat_prefix: str | None = None,
    bridge_sender: str | None = None,
    response_message_id: str | None = None,
    session_request_message_id: str | None = None,
    session_response_message_id: str | None = None,
    request_version: int | None = None,
) -> None:
    aliases = {
        "bridge_request_message_id": (
            bridge_request_message_id,
            "capability_request_script_event_id",
        ),
        "bridge_response_prefix": (bridge_response_prefix, "capability_response_chat_prefix"),
        "ui_chat_prefix": (ui_chat_prefix, "ui_chat_chunk_prefix"),
        "bridge_sender": (bridge_sender, "trusted_bridge_player_name"),
        "response_message_id": (response_message_id, "text_response_script_event_id"),
        "session_request_message_id": (
            session_request_message_id,
            "session_request_script_event_id",
        ),
        "session_response_message_id": (
            session_response_message_id,
            "session_response_script_event_id",
        ),
        "request_version": (request_version, "capability_request_schema_version"),
    }
    values: dict[str, object] = {
        "capability_request_script_event_id": capability_request_script_event_id,
        "capability_response_chat_prefix": capability_response_chat_prefix,
        "ui_chat_chunk_prefix": ui_chat_chunk_prefix,
        "session_request_chat_prefix": session_request_chat_prefix,
        "session_request_script_event_id": session_request_script_event_id,
        "session_response_script_event_id": session_response_script_event_id,
        "text_response_script_event_id": text_response_script_event_id,
        "approval_allow_chat_prefix": approval_allow_chat_prefix,
        "approval_deny_chat_prefix": approval_deny_chat_prefix,
        "trusted_bridge_player_name": trusted_bridge_player_name,
        "capability_request_schema_version": capability_request_schema_version,
        "session_schema_version": session_schema_version,
        "text_response_framing_version": text_response_framing_version,
        "ddui_persistence_version": ddui_persistence_version,
        "command_line_byte_budget": command_line_byte_budget,
        "upstream_max_content_code_points": upstream_max_content_code_points,
        "response_max_buffers": response_max_buffers,
        "response_max_chunks_per_message": response_max_chunks_per_message,
        "response_max_message_bytes": response_max_message_bytes,
        "response_max_total_buffer_bytes": response_max_total_buffer_bytes,
        "response_buffer_ttl_ms": response_buffer_ttl_ms,
        "session_response_max_command_bytes": session_response_max_command_bytes,
    }
    request_version_alias_used = request_version is not None
    for alias_name, (alias_value, semantic_name) in aliases.items():
        if alias_value is None:
            continue
        warnings.warn(
            f"McbewsV1Profile.{alias_name} is deprecated; use {semantic_name}",
            DeprecationWarning,
            stacklevel=2,
        )
        current = values[semantic_name]
        if current is not None and current != alias_value:
            raise ConfigurationError(
                f"{alias_name} conflicts with {semantic_name}; provide one name only"
            )
        values[semantic_name] = alias_value

    provided_values = {name: value for name, value in values.items() if value is not None}

    defaults: dict[str, object] = {
        "capability_request_script_event_id": _wire("capability_request_script_event_id"),
        "capability_response_chat_prefix": _wire("capability_response_chat_prefix"),
        "ui_chat_chunk_prefix": _wire("ui_chat_chunk_prefix"),
        "session_request_chat_prefix": _wire("session_request_chat_prefix"),
        "session_request_script_event_id": _wire("session_request_script_event_id"),
        "session_response_script_event_id": _wire("session_response_script_event_id"),
        "text_response_script_event_id": _wire("text_response_script_event_id"),
        "approval_allow_chat_prefix": _wire("approval_allow_chat_prefix"),
        "approval_deny_chat_prefix": _wire("approval_deny_chat_prefix"),
        "trusted_bridge_player_name": _wire("trusted_bridge_player_name"),
        "capability_request_schema_version": _version("capability_request_schema"),
        "session_schema_version": _version("session_schema"),
        "text_response_framing_version": _version("text_response_framing"),
        "ddui_persistence_version": _version("ddui_persistence"),
        "command_line_byte_budget": MCBEWS_V1_MANIFEST["limits"]["command_line_byte_budget"],
        "upstream_max_content_code_points": MCBEWS_V1_MANIFEST["limits"][
            "upstream_max_content_code_points"
        ],
        "response_max_buffers": MCBEWS_V1_MANIFEST["limits"]["response_max_buffers"],
        "response_max_chunks_per_message": MCBEWS_V1_MANIFEST["limits"][
            "response_max_chunks_per_message"
        ],
        "response_max_message_bytes": MCBEWS_V1_MANIFEST["limits"][
            "response_max_message_bytes"
        ],
        "response_max_total_buffer_bytes": MCBEWS_V1_MANIFEST["limits"][
            "response_max_total_buffer_bytes"
        ],
        "response_buffer_ttl_ms": MCBEWS_V1_MANIFEST["limits"]["response_buffer_ttl_ms"],
        "session_response_max_command_bytes": MCBEWS_V1_MANIFEST["limits"][
            "session_response_max_command_bytes"
        ],
    }
    for name, default in defaults.items():
        if values[name] is None:
            values[name] = default

    expected_capability_schema = defaults["capability_request_schema_version"]
    if values["capability_request_schema_version"] != expected_capability_schema:
        field_name = (
            "request_version"
            if request_version_alias_used
            else "capability_request_schema_version"
        )
        raise ConfigurationError(f"mcbews {field_name} must be {expected_capability_schema}")

    fixed_fields = (
        *defaults.keys(),
    )
    lowerable_limits = {"command_line_byte_budget", "session_response_max_command_bytes"}
    for name in fixed_fields:
        expected = cast(int, defaults[name])
        provided = provided_values.get(name, expected)
        if name in lowerable_limits:
            if type(provided) is not int or provided <= 0 or provided > expected:
                raise ConfigurationError(
                    f"mcbews {name} may only lower the manifest ceiling ({expected})"
                )
            values[name] = provided
            continue
        if provided != expected:
            raise ConfigurationError(f"mcbews {name} is fixed by the MCBEWS/1 manifest")
        values[name] = expected

    for name in (
        "upstream_max_content_code_points",
        "response_max_buffers",
        "response_max_chunks_per_message",
        "response_max_message_bytes",
        "response_max_total_buffer_bytes",
        "response_buffer_ttl_ms",
    ):
        value = values[name]
        if type(value) is not int or value <= 0:
            raise ConfigurationError(f"mcbews {name} must be a positive integer")
    for name in (
        "capability_request_script_event_id",
        "capability_response_chat_prefix",
        "ui_chat_chunk_prefix",
        "session_request_chat_prefix",
        "session_request_script_event_id",
        "session_response_script_event_id",
        "text_response_script_event_id",
        "approval_allow_chat_prefix",
        "approval_deny_chat_prefix",
        "trusted_bridge_player_name",
    ):
        value = values[name]
        if not isinstance(value, str) or not value.strip():
            raise ConfigurationError(f"mcbews {name} must be a non-empty string")
    _require_finite_non_negative_real(response_chunk_delay, "response_chunk_delay")
    _require_finite_non_negative_real(response_prelude_delay, "response_prelude_delay")
    protocol_line = MCBEWS_V1_MANIFEST.get("protocol_line")
    if not isinstance(protocol_line, str) or not protocol_line.strip():
        raise RuntimeError("MCBEWS/1 manifest protocol_line is invalid")
    object.__setattr__(self, "protocol_line", protocol_line)
    for name, value in values.items():
        object.__setattr__(self, name, value)
    object.__setattr__(self, "response_chunk_delay", response_chunk_delay)
    object.__setattr__(self, "response_prelude_delay", response_prelude_delay)

ToolPlayerMessage dataclass 

ToolPlayerMessage(channel: ToolPlayerChannel, sender: str, raw_message: str, session_request: SessionRequest | None = None, approval_decision: ApprovalDecision | None = None)

A control-channel message accepted only from the trusted ToolPlayer.

McbewsV1Delivery 

McbewsV1Delivery(outbound: McbeOutboundDelivery, *, profile: McbewsV1Profile = MCBEWS_V1, sleeper: Sleeper = asyncio.sleep)

Methods:

Source code in src/mcbe_ws_sdk/profiles/mcbews_v1/delivery.py 
18
19
20
21
22
23
24
25
26
27
def __init__(
    self,
    outbound: McbeOutboundDelivery,
    *,
    profile: McbewsV1Profile = MCBEWS_V1,
    sleeper: Sleeper = asyncio.sleep,
) -> None:
    self._outbound = outbound
    self._profile = profile
    self._sleep = sleeper

send_session_response async 

send_session_response(response: SessionResponse) -> None

Send one atomic correlated session response.

Session responses deliberately bypass generic scriptevent chunking. A result that does not fit is encoded as one SESSION_RESPONSE_TOO_LARGE error by :func:encode_session_response_command.

Source code in src/mcbe_ws_sdk/profiles/mcbews_v1/delivery.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
async def send_session_response(self, response: SessionResponse) -> None:
    """Send one atomic correlated session response.

    Session responses deliberately bypass generic scriptevent chunking. A
    result that does not fit is encoded as one ``SESSION_RESPONSE_TOO_LARGE``
    error by :func:`encode_session_response_command`.
    """

    payload = encode_session_response_command(
        response,
        self._outbound.flow,
        profile=self._profile,
    )
    await self._outbound.send_payload(payload, "session_resp")

ApprovalDecision

Bases: BaseModel

Typed approval decision carried by the ToolPlayer chat channel.

legacy=True is only used by the decoder for an id-only decision. Such a decision deliberately has no owner and must be resolved by a Host pending approval store; it is never treated as coming from the transport sender.

Attributes:

  • approved (bool) –

    Whether this decision approves the pending operation.

approved property 

approved: bool

Whether this decision approves the pending operation.

SessionError

Bases: BaseModel

Stable, bounded session error object.

SessionRequest

Bases: BaseModel

Validated session operation request from the trusted ToolPlayer.

SessionResponse

Bases: BaseModel

Atomic session response correlated to one :class:SessionRequest.

TextResponseChunk

Bases: BaseModel

One validated mcbews:text_resp frame.

TextResponseMessage

Bases: BaseModel

A complete, ordered text response after bounded reassembly.

TokenUsage

Bases: BaseModel

Compact token usage included only in a final text response frame.

UiChatMessage

Bases: BaseModel

A complete UI chat message with its originating conversation.

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 
197
198
199
200
201
202
203
204
205
@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)

classify_tool_player_message 

classify_tool_player_message(sender: str, message: str, profile: McbewsV1Profile = MCBEWS_V1) -> ToolPlayerMessage | None

Authenticate and classify one ToolPlayer chat message.

None means either that the sender is not the trusted ToolPlayer or the message is ordinary player chat. Malformed reserved messages raise a typed :class:ProtocolError; callers can log/drop them without passing them to a business hook.

Source code in src/mcbe_ws_sdk/profiles/mcbews_v1/classifier.py 
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
def classify_tool_player_message(
    sender: str,
    message: str,
    profile: McbewsV1Profile = MCBEWS_V1,
) -> ToolPlayerMessage | None:
    """Authenticate and classify one ToolPlayer chat message.

    ``None`` means either that the sender is not the trusted ToolPlayer or the
    message is ordinary player chat.  Malformed reserved messages raise a typed
    :class:`ProtocolError`; callers can log/drop them without passing them to a
    business hook.
    """

    if sender != profile.trusted_bridge_player_name:
        return None
    if message.startswith(f"{profile.capability_response_chat_prefix}|"):
        return ToolPlayerMessage("bridge", sender, message)
    if message.startswith(f"{profile.ui_chat_chunk_prefix}|"):
        return ToolPlayerMessage("ui_chat", sender, message)
    if message.startswith(f"{profile.session_request_chat_prefix}|"):
        request = decode_session_request_chat(message, profile=profile)
        return ToolPlayerMessage("session", sender, message, session_request=request)
    if message.startswith(f"{profile.approval_allow_chat_prefix}|") or message.startswith(
        f"{profile.approval_deny_chat_prefix}|"
    ):
        decision = decode_approval_decision(message, profile=profile)
        return ToolPlayerMessage("approval", sender, message, approval_decision=decision)
    return None

is_tool_player_channel_message 

is_tool_player_channel_message(message: str, profile: McbewsV1Profile = MCBEWS_V1) -> bool

Return whether a message uses a reserved MCBEWS/1 control prefix.

Source code in src/mcbe_ws_sdk/profiles/mcbews_v1/classifier.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
def is_tool_player_channel_message(
    message: str,
    profile: McbewsV1Profile = MCBEWS_V1,
) -> bool:
    """Return whether a message uses a reserved MCBEWS/1 control prefix."""

    return any(
        message.startswith(prefix)
        for prefix in (
            f"{profile.capability_response_chat_prefix}|",
            f"{profile.ui_chat_chunk_prefix}|",
            f"{profile.session_request_chat_prefix}|",
            f"{profile.approval_allow_chat_prefix}|",
            f"{profile.approval_deny_chat_prefix}|",
        )
    )

decode_approval_decision 

decode_approval_decision(message: str, profile: McbewsV1Profile = MCBEWS_V1) -> ApprovalDecision

Decode a typed or one-cycle legacy ToolPlayer approval decision.

Source code in src/mcbe_ws_sdk/profiles/mcbews_v1/codec.py 
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
def decode_approval_decision(
    message: str,
    profile: McbewsV1Profile = MCBEWS_V1,
) -> ApprovalDecision:
    """Decode a typed or one-cycle legacy ToolPlayer approval decision."""

    prefixes = (
        (f"{profile.approval_allow_chat_prefix}|", "approve"),
        (f"{profile.approval_deny_chat_prefix}|", "deny"),
    )
    for prefix, decision in prefixes:
        if not message.startswith(prefix):
            continue
        payload = message[len(prefix) :]
        try:
            value = json.loads(payload)
        except JSONDecodeError:
            value = payload
        return _decode_approval_payload(value, decision=decision)  # type: ignore[arg-type]
    raise ProtocolError("invalid approval decision prefix")

decode_session_request_chat 

decode_session_request_chat(message: str, profile: McbewsV1Profile = MCBEWS_V1) -> SessionRequest

Decode MCBEWS|SESSION|<json> into a validated DTO.

Source code in src/mcbe_ws_sdk/profiles/mcbews_v1/codec.py 
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
def decode_session_request_chat(
    message: str,
    profile: McbewsV1Profile = MCBEWS_V1,
) -> SessionRequest:
    """Decode ``MCBEWS|SESSION|<json>`` into a validated DTO."""

    prefix = f"{profile.session_request_chat_prefix}|"
    if not message.startswith(prefix):
        raise ProtocolError("invalid session request prefix")
    body = message[len(prefix) :]
    try:
        value = json.loads(body)
    except JSONDecodeError as exc:
        raise ProtocolError("invalid session request JSON") from exc
    if not isinstance(value, dict):
        raise ProtocolError("session request must be a JSON object")
    return _validate_model(SessionRequest, value)

decode_session_response_message 

decode_session_response_message(message: str, profile: McbewsV1Profile = MCBEWS_V1) -> SessionResponse

Decode an Addon mcbews:session_resp event body.

Source code in src/mcbe_ws_sdk/profiles/mcbews_v1/codec.py 
314
315
316
317
318
319
320
321
322
323
324
325
326
def decode_session_response_message(
    message: str,
    profile: McbewsV1Profile = MCBEWS_V1,
) -> SessionResponse:
    """Decode an Addon ``mcbews:session_resp`` event body."""

    try:
        value = json.loads(message)
    except JSONDecodeError as exc:
        raise ProtocolError("invalid session response JSON") from exc
    if not isinstance(value, dict):
        raise ProtocolError("session response must be a JSON object")
    return _validate_model(SessionResponse, value)

decode_text_response_frame 

decode_text_response_frame(value: str | dict[str, Any]) -> TextResponseChunk

Decode and validate one text response frame, rejecting unknown roles.

Source code in src/mcbe_ws_sdk/profiles/mcbews_v1/codec.py 
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
def decode_text_response_frame(value: str | dict[str, Any]) -> TextResponseChunk:
    """Decode and validate one text response frame, rejecting unknown roles."""

    if isinstance(value, str):
        try:
            value = json.loads(value)
        except JSONDecodeError as exc:
            raise ProtocolError("invalid text response JSON") from exc
    if not isinstance(value, dict):
        raise ProtocolError("text response frame must be a JSON object")
    if value.get("r") not in {"user", "assistant", "approval"}:
        raise ProtocolError(f"UNKNOWN_TEXT_ROLE: {value.get('r')!r}")
    try:
        return TextResponseChunk.model_validate(value)
    except ValidationError as exc:
        message = str(exc)
        if "role" in message:
            message = f"UNKNOWN_TEXT_ROLE: {message}"
        raise ProtocolError(message) from exc

encode_approval_decision 

encode_approval_decision(decision: ApprovalDecision, profile: McbewsV1Profile = MCBEWS_V1) -> str

Encode a typed approval decision for the trusted ToolPlayer channel.

Source code in src/mcbe_ws_sdk/profiles/mcbews_v1/codec.py 
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
def encode_approval_decision(
    decision: ApprovalDecision,
    profile: McbewsV1Profile = MCBEWS_V1,
) -> str:
    """Encode a typed approval decision for the trusted ToolPlayer channel."""

    prefix = (
        profile.approval_allow_chat_prefix
        if decision.approved
        else profile.approval_deny_chat_prefix
    )
    if decision.legacy:
        return f"{prefix}|{decision.approval_id}"
    body = decision.model_dump(
        by_alias=True,
        exclude_none=True,
        exclude={"legacy", "decision"},
    )
    return f"{prefix}|{json.dumps(body, ensure_ascii=False, separators=(',', ':'))}"

encode_session_request_chat 

encode_session_request_chat(request: SessionRequest, profile: McbewsV1Profile = MCBEWS_V1) -> str

Encode the atomic session request chat envelope.

Source code in src/mcbe_ws_sdk/profiles/mcbews_v1/codec.py 
214
215
216
217
218
219
220
221
222
223
224
225
def encode_session_request_chat(
    request: SessionRequest,
    profile: McbewsV1Profile = MCBEWS_V1,
) -> str:
    """Encode the atomic session request chat envelope."""

    body = json.dumps(
        request.model_dump(by_alias=True, exclude_none=True),
        ensure_ascii=False,
        separators=(",", ":"),
    )
    return f"{profile.session_request_chat_prefix}|{body}"

encode_session_response_command 

encode_session_response_command(response: SessionResponse, flow: FlowControlMiddleware, profile: McbewsV1Profile = MCBEWS_V1) -> str

Encode one atomic mcbews:session_resp commandRequest payload.

A large successful result is replaced by a correlated structured error. No generic chunker is called, so a receiver never sees a fragment that looks like a complete JSON response.

Source code in src/mcbe_ws_sdk/profiles/mcbews_v1/codec.py 
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
def encode_session_response_command(
    response: SessionResponse,
    flow: FlowControlMiddleware,
    profile: McbewsV1Profile = MCBEWS_V1,
) -> str:
    """Encode one atomic ``mcbews:session_resp`` commandRequest payload.

    A large successful result is replaced by a correlated structured error.  No
    generic chunker is called, so a receiver never sees a fragment that looks
    like a complete JSON response.
    """

    command = (
        f"scriptevent {profile.session_response_script_event_id} "
        f"{encode_session_response_json(response)}"
    )
    def atomic_payload(raw_command: str) -> str:
        if len(raw_command.encode("utf-8")) > profile.session_response_max_command_bytes:
            raise FrameTooLargeError(
                "session response command exceeds the MCBEWS/1 atomic command budget"
            )
        return flow.chunk_raw_command(raw_command)[0]

    try:
        return atomic_payload(command)
    except FrameTooLargeError:
        fallback = _session_response_error(
            response,
            request_id=response.request_id,
            action=response.action,
        )
        fallback_command = (
            f"scriptevent {profile.session_response_script_event_id} "
            f"{encode_session_response_json(fallback)}"
        )
        try:
            return atomic_payload(fallback_command)
        except FrameTooLargeError as exc:
            raise FrameTooLargeError(
                "SESSION_RESPONSE_TOO_LARGE error cannot fit the configured atomic frame budget"
            ) from exc

load_manifest 

load_manifest() -> Mapping[str, Any]

Load the immutable MCBEWS/1 manifest from the installed package.

Source code in src/mcbe_ws_sdk/profiles/mcbews_v1/manifest.py 
35
36
37
38
def load_manifest() -> Mapping[str, Any]:
    """Load the immutable MCBEWS/1 manifest from the installed package."""

    return cast(Mapping[str, Any], _freeze(_load_json("manifest.json")))

load_wire_vectors 

load_wire_vectors() -> Mapping[str, Any]

Load the immutable executable MCBEWS/1 conformance vectors.

Source code in src/mcbe_ws_sdk/profiles/mcbews_v1/manifest.py 
41
42
43
44
def load_wire_vectors() -> Mapping[str, Any]:
    """Load the immutable executable MCBEWS/1 conformance vectors."""

    return cast(Mapping[str, Any], _freeze(_load_json("vectors.json")))

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.

  • AddonControlHook –

    Optional hook for authenticated session/approval ToolPlayer DTOs.

  • ConnectionHook –

    Lifecycle + protocol hooks the host injects into the gateway.

  • LegacyUiChatHook –

    Pre-0.2.0 UI callback shape used only through an explicit adapter.

  • LegacyUiChatHookAdapter –

    Adapt a legacy three-argument UI hook to the typed callback boundary.

  • 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)

AddonControlHook

Bases: Protocol

Optional hook for authenticated session/approval ToolPlayer DTOs.

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 
33
34
35
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 
37
38
39
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 
41
42
43
44
45
46
47
48
49
50
51
52
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, message: UiChatMessage) -> None

Fired when a fragmented UI_CHAT message is fully reassembled.

Source code in src/mcbe_ws_sdk/gateway/hook.py 
54
55
56
57
async def on_ui_chat_reassembled(self, state: ConnectionState, message: UiChatMessage) -> 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 
59
60
61
62
63
64
65
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 
67
68
69
async def on_error(self, state: ConnectionState, error: MinecraftErrorFrame) -> None:
    """Inbound typed ``error`` frame."""
    ...

LegacyUiChatHook

Bases: Protocol

Pre-0.2.0 UI callback shape used only through an explicit adapter.

LegacyUiChatHookAdapter 

LegacyUiChatHookAdapter(legacy_hook: LegacyUiChatHook)

Adapt a legacy three-argument UI hook to the typed callback boundary.

Source code in src/mcbe_ws_sdk/gateway/hook.py 
125
126
def __init__(self, legacy_hook: LegacyUiChatHook) -> None:
    self._legacy_hook = legacy_hook

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, legacy_ui_chat_hook: LegacyUiChatHook | 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 
 72
 73
 74
 75
 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
def __init__(
    self,
    *,
    settings: GatewaySettings | None = None,
    hook: ConnectionHook | None = None,
    legacy_ui_chat_hook: LegacyUiChatHook | 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._legacy_ui_chat_hook = (
        LegacyUiChatHookAdapter(legacy_ui_chat_hook)
        if legacy_ui_chat_hook is not None
        else None
    )
    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 
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
161
162
163
164
165
166
167
168
169
170
171
172
173
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 
175
176
177
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.

  • LegacyUiChatHook –

    Pre-0.2.0 UI callback shape used only through an explicit adapter.

  • LegacyUiChatHookAdapter –

    Adapt a legacy three-argument UI hook to the typed callback boundary.

  • AddonControlHook –

    Optional hook for authenticated session/approval ToolPlayer DTOs.

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 
33
34
35
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 
37
38
39
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 
41
42
43
44
45
46
47
48
49
50
51
52
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, message: UiChatMessage) -> None

Fired when a fragmented UI_CHAT message is fully reassembled.

Source code in src/mcbe_ws_sdk/gateway/hook.py 
54
55
56
57
async def on_ui_chat_reassembled(self, state: ConnectionState, message: UiChatMessage) -> 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 
59
60
61
62
63
64
65
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 
67
68
69
async def on_error(self, state: ConnectionState, error: MinecraftErrorFrame) -> None:
    """Inbound typed ``error`` frame."""
    ...

NoOpHook

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

LegacyUiChatHook

Bases: Protocol

Pre-0.2.0 UI callback shape used only through an explicit adapter.

LegacyUiChatHookAdapter 

LegacyUiChatHookAdapter(legacy_hook: LegacyUiChatHook)

Adapt a legacy three-argument UI hook to the typed callback boundary.

Source code in src/mcbe_ws_sdk/gateway/hook.py 
125
126
def __init__(self, legacy_hook: LegacyUiChatHook) -> None:
    self._legacy_hook = legacy_hook

AddonControlHook

Bases: Protocol

Optional hook for authenticated session/approval ToolPlayer DTOs.

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, legacy_ui_chat_hook: LegacyUiChatHook | 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 
 72
 73
 74
 75
 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
def __init__(
    self,
    *,
    settings: GatewaySettings | None = None,
    hook: ConnectionHook | None = None,
    legacy_ui_chat_hook: LegacyUiChatHook | 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._legacy_ui_chat_hook = (
        LegacyUiChatHookAdapter(legacy_ui_chat_hook)
        if legacy_ui_chat_hook is not None
        else None
    )
    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 
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
161
162
163
164
165
166
167
168
169
170
171
172
173
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 
175
176
177
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 
71
72
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 
78
79
80
81
82
83
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 
85
86
87
88
89
90
91
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 
 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
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
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
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,
        channel="capability",
        payload_bytes=payload_bytes,
    )
    logger.debug(
        "bridge_request_outbound",
        connection_id=str(connection_id),
        request_id=request.request_id,
        capability=capability,
        command_bytes=len(command.encode("utf-8")),
        timeout_seconds=self._timeout_seconds,
    )
    try:
        try:
            await send_command(command)
        except Exception as exc:
            # Fail-fast before waiting: host rejected the outbound frame
            # (e.g. FrameTooLarge) or the connection is gone. Waiting for a
            # RESP that can never arrive only produces a misleading timeout.
            logger.warning(
                "bridge_request_send_failed",
                connection_id=str(connection_id),
                request_id=request.request_id,
                channel="capability",
                payload_bytes=payload_bytes,
                error_type=type(exc).__name__,
            )
            raise
        logger.info(
            "bridge_request_sent",
            connection_id=str(connection_id),
            request_id=request.request_id,
            channel="capability",
            payload_bytes=payload_bytes,
        )
        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,
                channel="capability",
                payload_bytes=payload_bytes,
            )
            raise BridgeTimeoutError(request.request_id) from exc
        logger.info(
            "bridge_request_resolved",
            connection_id=str(connection_id),
            request_id=request.request_id,
            channel="capability",
            result_type=type(result).__name__,
        )
        logger.debug(
            "bridge_request_resolved_payload",
            connection_id=str(connection_id),
            request_id=request.request_id,
            capability=capability,
            result_type=type(result).__name__,
        )
        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 
179
180
181
182
183
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.trusted_bridge_player_name and message.startswith(
        f"{self._profile.capability_response_chat_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 
185
186
187
188
189
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.trusted_bridge_player_name and message.startswith(
        f"{self._profile.ui_chat_chunk_prefix}|"
    )

is_tool_player_channel_message 

is_tool_player_channel_message(message: str) -> bool

Return whether message is reserved for a ToolPlayer channel.

Source code in src/mcbe_ws_sdk/addon/service.py 
191
192
193
194
def is_tool_player_channel_message(self, message: str) -> bool:
    """Return whether *message* is reserved for a ToolPlayer channel."""

    return is_tool_player_channel_message(message, profile=self._profile)

classify_tool_player_message 

classify_tool_player_message(sender: str, message: str) -> ToolPlayerMessage | None

Authenticate and classify a reserved ToolPlayer message.

Source code in src/mcbe_ws_sdk/addon/service.py 
196
197
198
199
def classify_tool_player_message(self, sender: str, message: str) -> ToolPlayerMessage | None:
    """Authenticate and classify a reserved ToolPlayer message."""

    return classify_tool_player_message(sender, message, profile=self._profile)

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 
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
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):
        # Keep INFO body-free: these frames 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),
            channel="capability",
            message_bytes=len(message.encode("utf-8")),
        )
        session = self._sessions.get(connection_id)
        if session is None:
            logger.warning(
                "bridge_chat_no_session",
                connection_id=str(connection_id),
                channel="capability",
            )
            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),
            channel="ui_chat",
            message_bytes=len(message.encode("utf-8")),
        )
        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),
                channel="ui_chat",
                message_bytes=len(ui_message.message.encode("utf-8")),
            )
            await self._invoke_ui_chat_callback(connection_id, ui_message)
        return AddonMessageResult(
            handled=True,
            ui_chunk=ui_chunk,
            ui_message=ui_message,
        )

    if self.is_tool_player_channel_message(message):
        # Session and approval frames are decoded only after the trusted
        # sender gate.  The facade uses this branch to deliver typed control
        # messages to the Host hook; they never fall through as player chat.
        control_message = self.classify_tool_player_message(sender, message)
        if control_message is None:
            return AddonMessageResult(handled=True)
        return AddonMessageResult(handled=True, control_message=control_message)

    return AddonMessageResult(handled=False)

set_ui_chat_callback 

set_ui_chat_callback(callback: UiChatCallback) -> None

Register the typed UI chat message callback.

Source code in src/mcbe_ws_sdk/addon/service.py 
261
262
263
264
265
266
def set_ui_chat_callback(
    self,
    callback: UiChatCallback,
) -> None:
    """Register the typed UI chat message callback."""
    self._ui_chat_callback = callback

set_legacy_ui_chat_callback 

set_legacy_ui_chat_callback(callback: LegacyUiChatCallback) -> None

Register a legacy callback through an explicit migration adapter.

Source code in src/mcbe_ws_sdk/addon/service.py 
268
269
270
def set_legacy_ui_chat_callback(self, callback: LegacyUiChatCallback) -> None:
    """Register a legacy callback through an explicit migration adapter."""
    self._ui_chat_callback = LegacyUiChatCallbackAdapter(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 
279
280
281
282
283
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 
296
297
298
299
300
301
302
303
304
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

LegacyUiChatCallbackAdapter 

LegacyUiChatCallbackAdapter(callback: LegacyUiChatCallback)

Adapt the pre-0.2.0 three-argument callback to the typed DTO callback.

Source code in src/mcbe_ws_sdk/addon/service.py 
52
53
def __init__(self, callback: LegacyUiChatCallback) -> None:
    self._callback = callback

ToolPlayerMessage dataclass 

ToolPlayerMessage(channel: ToolPlayerChannel, sender: str, raw_message: str, session_request: SessionRequest | None = None, approval_decision: ApprovalDecision | None = None)

A control-channel message accepted only from the trusted ToolPlayer.

ApprovalDecision

Bases: BaseModel

Typed approval decision carried by the ToolPlayer chat channel.

legacy=True is only used by the decoder for an id-only decision. Such a decision deliberately has no owner and must be resolved by a Host pending approval store; it is never treated as coming from the transport sender.

Attributes:

  • approved (bool) –

    Whether this decision approves the pending operation.

approved property 

approved: bool

Whether this decision approves the pending operation.

SessionRequest

Bases: BaseModel

Validated session operation request from the trusted ToolPlayer.

SessionResponse

Bases: BaseModel

Atomic session response correlated to one :class:SessionRequest.

TextResponseChunk

Bases: BaseModel

One validated mcbews:text_resp frame.

TextResponseMessage

Bases: BaseModel

A complete, ordered text response after bounded reassembly.

UiChatMessage

Bases: BaseModel

A complete UI chat message with its originating conversation.

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:

LegacyUiChatCallbackAdapter 

LegacyUiChatCallbackAdapter(callback: LegacyUiChatCallback)

Adapt the pre-0.2.0 three-argument callback to the typed DTO callback.

Source code in src/mcbe_ws_sdk/addon/service.py 
52
53
def __init__(self, callback: LegacyUiChatCallback) -> None:
    self._callback = callback

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 
71
72
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 
78
79
80
81
82
83
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 
85
86
87
88
89
90
91
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 
 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
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
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
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,
        channel="capability",
        payload_bytes=payload_bytes,
    )
    logger.debug(
        "bridge_request_outbound",
        connection_id=str(connection_id),
        request_id=request.request_id,
        capability=capability,
        command_bytes=len(command.encode("utf-8")),
        timeout_seconds=self._timeout_seconds,
    )
    try:
        try:
            await send_command(command)
        except Exception as exc:
            # Fail-fast before waiting: host rejected the outbound frame
            # (e.g. FrameTooLarge) or the connection is gone. Waiting for a
            # RESP that can never arrive only produces a misleading timeout.
            logger.warning(
                "bridge_request_send_failed",
                connection_id=str(connection_id),
                request_id=request.request_id,
                channel="capability",
                payload_bytes=payload_bytes,
                error_type=type(exc).__name__,
            )
            raise
        logger.info(
            "bridge_request_sent",
            connection_id=str(connection_id),
            request_id=request.request_id,
            channel="capability",
            payload_bytes=payload_bytes,
        )
        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,
                channel="capability",
                payload_bytes=payload_bytes,
            )
            raise BridgeTimeoutError(request.request_id) from exc
        logger.info(
            "bridge_request_resolved",
            connection_id=str(connection_id),
            request_id=request.request_id,
            channel="capability",
            result_type=type(result).__name__,
        )
        logger.debug(
            "bridge_request_resolved_payload",
            connection_id=str(connection_id),
            request_id=request.request_id,
            capability=capability,
            result_type=type(result).__name__,
        )
        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 
179
180
181
182
183
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.trusted_bridge_player_name and message.startswith(
        f"{self._profile.capability_response_chat_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 
185
186
187
188
189
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.trusted_bridge_player_name and message.startswith(
        f"{self._profile.ui_chat_chunk_prefix}|"
    )
is_tool_player_channel_message 
is_tool_player_channel_message(message: str) -> bool

Return whether message is reserved for a ToolPlayer channel.

Source code in src/mcbe_ws_sdk/addon/service.py 
191
192
193
194
def is_tool_player_channel_message(self, message: str) -> bool:
    """Return whether *message* is reserved for a ToolPlayer channel."""

    return is_tool_player_channel_message(message, profile=self._profile)
classify_tool_player_message 
classify_tool_player_message(sender: str, message: str) -> ToolPlayerMessage | None

Authenticate and classify a reserved ToolPlayer message.

Source code in src/mcbe_ws_sdk/addon/service.py 
196
197
198
199
def classify_tool_player_message(self, sender: str, message: str) -> ToolPlayerMessage | None:
    """Authenticate and classify a reserved ToolPlayer message."""

    return classify_tool_player_message(sender, message, profile=self._profile)
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 
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
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):
        # Keep INFO body-free: these frames 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),
            channel="capability",
            message_bytes=len(message.encode("utf-8")),
        )
        session = self._sessions.get(connection_id)
        if session is None:
            logger.warning(
                "bridge_chat_no_session",
                connection_id=str(connection_id),
                channel="capability",
            )
            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),
            channel="ui_chat",
            message_bytes=len(message.encode("utf-8")),
        )
        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),
                channel="ui_chat",
                message_bytes=len(ui_message.message.encode("utf-8")),
            )
            await self._invoke_ui_chat_callback(connection_id, ui_message)
        return AddonMessageResult(
            handled=True,
            ui_chunk=ui_chunk,
            ui_message=ui_message,
        )

    if self.is_tool_player_channel_message(message):
        # Session and approval frames are decoded only after the trusted
        # sender gate.  The facade uses this branch to deliver typed control
        # messages to the Host hook; they never fall through as player chat.
        control_message = self.classify_tool_player_message(sender, message)
        if control_message is None:
            return AddonMessageResult(handled=True)
        return AddonMessageResult(handled=True, control_message=control_message)

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

Register the typed UI chat message callback.

Source code in src/mcbe_ws_sdk/addon/service.py 
261
262
263
264
265
266
def set_ui_chat_callback(
    self,
    callback: UiChatCallback,
) -> None:
    """Register the typed UI chat message callback."""
    self._ui_chat_callback = callback
set_legacy_ui_chat_callback 
set_legacy_ui_chat_callback(callback: LegacyUiChatCallback) -> None

Register a legacy callback through an explicit migration adapter.

Source code in src/mcbe_ws_sdk/addon/service.py 
268
269
270
def set_legacy_ui_chat_callback(self, callback: LegacyUiChatCallback) -> None:
    """Register a legacy callback through an explicit migration adapter."""
    self._ui_chat_callback = LegacyUiChatCallbackAdapter(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 
279
280
281
282
283
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 
296
297
298
299
300
301
302
303
304
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 
197
198
199
200
201
202
203
204
205
@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 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
@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 
152
153
154
155
156
157
158
159
160
@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 
162
163
164
165
@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 
181
182
183
184
@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 
197
198
199
200
201
202
203
204
205
@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

Prepare tellraw body text before embedding in the rawtext JSON value.

Quote/backslash escaping is left to json.dumps. Plain % is left unchanged: Bedrock rawtext text components render a single percent literally, and doubling it surfaces as %% in chat (e.g. context usage tips like 20.0%).

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

    Quote/backslash escaping is left to ``json.dumps``. Plain ``%`` is left
    unchanged: Bedrock ``rawtext`` ``text`` components render a single percent
    literally, and doubling it surfaces as ``%%`` in chat (e.g. context usage
    tips like ``20.0%``).
    """
    return message

Profiles

mcbe_ws_sdk.profiles

MCBEWS/1 protocol profile exports.

MCBEWS/1 is the sole runtime profile. AddonBridgeProfile is retained as a deprecated type alias/re-export for source compatibility; it is not a runtime extension seam and SDK settings accept the concrete profile only.

Modules:

  • mcbews_v1 –

    MCBEWS/1 protocol implementation and canonical assets.

Classes:

McbewsV1Profile dataclass 

McbewsV1Profile(*, capability_request_script_event_id: str | None = None, capability_response_chat_prefix: str | None = None, ui_chat_chunk_prefix: str | None = None, session_request_chat_prefix: str | None = None, session_request_script_event_id: str | None = None, session_response_script_event_id: str | None = None, text_response_script_event_id: str | None = None, approval_allow_chat_prefix: str | None = None, approval_deny_chat_prefix: str | None = None, trusted_bridge_player_name: str | None = None, capability_request_schema_version: int | None = None, session_schema_version: int | None = None, text_response_framing_version: int | None = None, ddui_persistence_version: int | None = None, command_line_byte_budget: int | None = None, upstream_max_content_code_points: int | None = None, response_max_buffers: int | None = None, response_max_chunks_per_message: int | None = None, response_max_message_bytes: int | None = None, response_max_total_buffer_bytes: int | None = None, response_buffer_ttl_ms: int | None = None, session_response_max_command_bytes: int | None = None, response_chunk_delay: float = 0.15, response_prelude_delay: float = 0.5, bridge_request_message_id: str | None = None, bridge_response_prefix: str | None = None, ui_chat_prefix: str | None = None, bridge_sender: str | None = None, response_message_id: str | None = None, session_request_message_id: str | None = None, session_response_message_id: str | None = None, request_version: int | None = None)

Concrete MCBEWS/1 wire profile.

The semantic field names are the protocol authority. Wire identifiers, schema versions and safety bounds come from the installed manifest and cannot be replaced by a runtime profile seam. The empirical command budget may be lowered for a deployment and response delays remain operational knobs. Historical names are accepted for one migration cycle only when they still select the manifest value.

Source code in src/mcbe_ws_sdk/profiles/mcbews_v1/profile.py 
 70
 71
 72
 73
 74
 75
 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
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
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
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
def __init__(
    self,
    *,
    capability_request_script_event_id: str | None = None,
    capability_response_chat_prefix: str | None = None,
    ui_chat_chunk_prefix: str | None = None,
    session_request_chat_prefix: str | None = None,
    session_request_script_event_id: str | None = None,
    session_response_script_event_id: str | None = None,
    text_response_script_event_id: str | None = None,
    approval_allow_chat_prefix: str | None = None,
    approval_deny_chat_prefix: str | None = None,
    trusted_bridge_player_name: str | None = None,
    capability_request_schema_version: int | None = None,
    session_schema_version: int | None = None,
    text_response_framing_version: int | None = None,
    ddui_persistence_version: int | None = None,
    command_line_byte_budget: int | None = None,
    upstream_max_content_code_points: int | None = None,
    response_max_buffers: int | None = None,
    response_max_chunks_per_message: int | None = None,
    response_max_message_bytes: int | None = None,
    response_max_total_buffer_bytes: int | None = None,
    response_buffer_ttl_ms: int | None = None,
    session_response_max_command_bytes: int | None = None,
    response_chunk_delay: float = 0.15,
    response_prelude_delay: float = 0.5,
    # Deprecated aliases.  They are deliberately not stored as independent
    # values, so alias and semantic field parity cannot drift.
    bridge_request_message_id: str | None = None,
    bridge_response_prefix: str | None = None,
    ui_chat_prefix: str | None = None,
    bridge_sender: str | None = None,
    response_message_id: str | None = None,
    session_request_message_id: str | None = None,
    session_response_message_id: str | None = None,
    request_version: int | None = None,
) -> None:
    aliases = {
        "bridge_request_message_id": (
            bridge_request_message_id,
            "capability_request_script_event_id",
        ),
        "bridge_response_prefix": (bridge_response_prefix, "capability_response_chat_prefix"),
        "ui_chat_prefix": (ui_chat_prefix, "ui_chat_chunk_prefix"),
        "bridge_sender": (bridge_sender, "trusted_bridge_player_name"),
        "response_message_id": (response_message_id, "text_response_script_event_id"),
        "session_request_message_id": (
            session_request_message_id,
            "session_request_script_event_id",
        ),
        "session_response_message_id": (
            session_response_message_id,
            "session_response_script_event_id",
        ),
        "request_version": (request_version, "capability_request_schema_version"),
    }
    values: dict[str, object] = {
        "capability_request_script_event_id": capability_request_script_event_id,
        "capability_response_chat_prefix": capability_response_chat_prefix,
        "ui_chat_chunk_prefix": ui_chat_chunk_prefix,
        "session_request_chat_prefix": session_request_chat_prefix,
        "session_request_script_event_id": session_request_script_event_id,
        "session_response_script_event_id": session_response_script_event_id,
        "text_response_script_event_id": text_response_script_event_id,
        "approval_allow_chat_prefix": approval_allow_chat_prefix,
        "approval_deny_chat_prefix": approval_deny_chat_prefix,
        "trusted_bridge_player_name": trusted_bridge_player_name,
        "capability_request_schema_version": capability_request_schema_version,
        "session_schema_version": session_schema_version,
        "text_response_framing_version": text_response_framing_version,
        "ddui_persistence_version": ddui_persistence_version,
        "command_line_byte_budget": command_line_byte_budget,
        "upstream_max_content_code_points": upstream_max_content_code_points,
        "response_max_buffers": response_max_buffers,
        "response_max_chunks_per_message": response_max_chunks_per_message,
        "response_max_message_bytes": response_max_message_bytes,
        "response_max_total_buffer_bytes": response_max_total_buffer_bytes,
        "response_buffer_ttl_ms": response_buffer_ttl_ms,
        "session_response_max_command_bytes": session_response_max_command_bytes,
    }
    request_version_alias_used = request_version is not None
    for alias_name, (alias_value, semantic_name) in aliases.items():
        if alias_value is None:
            continue
        warnings.warn(
            f"McbewsV1Profile.{alias_name} is deprecated; use {semantic_name}",
            DeprecationWarning,
            stacklevel=2,
        )
        current = values[semantic_name]
        if current is not None and current != alias_value:
            raise ConfigurationError(
                f"{alias_name} conflicts with {semantic_name}; provide one name only"
            )
        values[semantic_name] = alias_value

    provided_values = {name: value for name, value in values.items() if value is not None}

    defaults: dict[str, object] = {
        "capability_request_script_event_id": _wire("capability_request_script_event_id"),
        "capability_response_chat_prefix": _wire("capability_response_chat_prefix"),
        "ui_chat_chunk_prefix": _wire("ui_chat_chunk_prefix"),
        "session_request_chat_prefix": _wire("session_request_chat_prefix"),
        "session_request_script_event_id": _wire("session_request_script_event_id"),
        "session_response_script_event_id": _wire("session_response_script_event_id"),
        "text_response_script_event_id": _wire("text_response_script_event_id"),
        "approval_allow_chat_prefix": _wire("approval_allow_chat_prefix"),
        "approval_deny_chat_prefix": _wire("approval_deny_chat_prefix"),
        "trusted_bridge_player_name": _wire("trusted_bridge_player_name"),
        "capability_request_schema_version": _version("capability_request_schema"),
        "session_schema_version": _version("session_schema"),
        "text_response_framing_version": _version("text_response_framing"),
        "ddui_persistence_version": _version("ddui_persistence"),
        "command_line_byte_budget": MCBEWS_V1_MANIFEST["limits"]["command_line_byte_budget"],
        "upstream_max_content_code_points": MCBEWS_V1_MANIFEST["limits"][
            "upstream_max_content_code_points"
        ],
        "response_max_buffers": MCBEWS_V1_MANIFEST["limits"]["response_max_buffers"],
        "response_max_chunks_per_message": MCBEWS_V1_MANIFEST["limits"][
            "response_max_chunks_per_message"
        ],
        "response_max_message_bytes": MCBEWS_V1_MANIFEST["limits"][
            "response_max_message_bytes"
        ],
        "response_max_total_buffer_bytes": MCBEWS_V1_MANIFEST["limits"][
            "response_max_total_buffer_bytes"
        ],
        "response_buffer_ttl_ms": MCBEWS_V1_MANIFEST["limits"]["response_buffer_ttl_ms"],
        "session_response_max_command_bytes": MCBEWS_V1_MANIFEST["limits"][
            "session_response_max_command_bytes"
        ],
    }
    for name, default in defaults.items():
        if values[name] is None:
            values[name] = default

    expected_capability_schema = defaults["capability_request_schema_version"]
    if values["capability_request_schema_version"] != expected_capability_schema:
        field_name = (
            "request_version"
            if request_version_alias_used
            else "capability_request_schema_version"
        )
        raise ConfigurationError(f"mcbews {field_name} must be {expected_capability_schema}")

    fixed_fields = (
        *defaults.keys(),
    )
    lowerable_limits = {"command_line_byte_budget", "session_response_max_command_bytes"}
    for name in fixed_fields:
        expected = cast(int, defaults[name])
        provided = provided_values.get(name, expected)
        if name in lowerable_limits:
            if type(provided) is not int or provided <= 0 or provided > expected:
                raise ConfigurationError(
                    f"mcbews {name} may only lower the manifest ceiling ({expected})"
                )
            values[name] = provided
            continue
        if provided != expected:
            raise ConfigurationError(f"mcbews {name} is fixed by the MCBEWS/1 manifest")
        values[name] = expected

    for name in (
        "upstream_max_content_code_points",
        "response_max_buffers",
        "response_max_chunks_per_message",
        "response_max_message_bytes",
        "response_max_total_buffer_bytes",
        "response_buffer_ttl_ms",
    ):
        value = values[name]
        if type(value) is not int or value <= 0:
            raise ConfigurationError(f"mcbews {name} must be a positive integer")
    for name in (
        "capability_request_script_event_id",
        "capability_response_chat_prefix",
        "ui_chat_chunk_prefix",
        "session_request_chat_prefix",
        "session_request_script_event_id",
        "session_response_script_event_id",
        "text_response_script_event_id",
        "approval_allow_chat_prefix",
        "approval_deny_chat_prefix",
        "trusted_bridge_player_name",
    ):
        value = values[name]
        if not isinstance(value, str) or not value.strip():
            raise ConfigurationError(f"mcbews {name} must be a non-empty string")
    _require_finite_non_negative_real(response_chunk_delay, "response_chunk_delay")
    _require_finite_non_negative_real(response_prelude_delay, "response_prelude_delay")
    protocol_line = MCBEWS_V1_MANIFEST.get("protocol_line")
    if not isinstance(protocol_line, str) or not protocol_line.strip():
        raise RuntimeError("MCBEWS/1 manifest protocol_line is invalid")
    object.__setattr__(self, "protocol_line", protocol_line)
    for name, value in values.items():
        object.__setattr__(self, name, value)
    object.__setattr__(self, "response_chunk_delay", response_chunk_delay)
    object.__setattr__(self, "response_prelude_delay", response_prelude_delay)

mcbews_v1

MCBEWS/1 protocol implementation and canonical assets.

Modules:

Classes:

Functions:

ToolPlayerMessage dataclass 

ToolPlayerMessage(channel: ToolPlayerChannel, sender: str, raw_message: str, session_request: SessionRequest | None = None, approval_decision: ApprovalDecision | None = None)

A control-channel message accepted only from the trusted ToolPlayer.

AddonBridgeRequest

Bases: BaseModel

Capability request sent to the reference Addon.

ApprovalDecision

Bases: BaseModel

Typed approval decision carried by the ToolPlayer chat channel.

legacy=True is only used by the decoder for an id-only decision. Such a decision deliberately has no owner and must be resolved by a Host pending approval store; it is never treated as coming from the transport sender.

Attributes:

  • approved (bool) –

    Whether this decision approves the pending operation.

approved property 
approved: bool

Whether this decision approves the pending operation.

SessionError

Bases: BaseModel

Stable, bounded session error object.

SessionRequest

Bases: BaseModel

Validated session operation request from the trusted ToolPlayer.

SessionResponse

Bases: BaseModel

Atomic session response correlated to one :class:SessionRequest.

TextResponseChunk

Bases: BaseModel

One validated mcbews:text_resp frame.

TextResponseMessage

Bases: BaseModel

A complete, ordered text response after bounded reassembly.

TokenUsage

Bases: BaseModel

Compact token usage included only in a final text response frame.

UiChatMessage

Bases: BaseModel

A complete UI chat message with its originating conversation.

McbewsV1Profile dataclass 

McbewsV1Profile(*, capability_request_script_event_id: str | None = None, capability_response_chat_prefix: str | None = None, ui_chat_chunk_prefix: str | None = None, session_request_chat_prefix: str | None = None, session_request_script_event_id: str | None = None, session_response_script_event_id: str | None = None, text_response_script_event_id: str | None = None, approval_allow_chat_prefix: str | None = None, approval_deny_chat_prefix: str | None = None, trusted_bridge_player_name: str | None = None, capability_request_schema_version: int | None = None, session_schema_version: int | None = None, text_response_framing_version: int | None = None, ddui_persistence_version: int | None = None, command_line_byte_budget: int | None = None, upstream_max_content_code_points: int | None = None, response_max_buffers: int | None = None, response_max_chunks_per_message: int | None = None, response_max_message_bytes: int | None = None, response_max_total_buffer_bytes: int | None = None, response_buffer_ttl_ms: int | None = None, session_response_max_command_bytes: int | None = None, response_chunk_delay: float = 0.15, response_prelude_delay: float = 0.5, bridge_request_message_id: str | None = None, bridge_response_prefix: str | None = None, ui_chat_prefix: str | None = None, bridge_sender: str | None = None, response_message_id: str | None = None, session_request_message_id: str | None = None, session_response_message_id: str | None = None, request_version: int | None = None)

Concrete MCBEWS/1 wire profile.

The semantic field names are the protocol authority. Wire identifiers, schema versions and safety bounds come from the installed manifest and cannot be replaced by a runtime profile seam. The empirical command budget may be lowered for a deployment and response delays remain operational knobs. Historical names are accepted for one migration cycle only when they still select the manifest value.

Source code in src/mcbe_ws_sdk/profiles/mcbews_v1/profile.py 
 70
 71
 72
 73
 74
 75
 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
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
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
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
def __init__(
    self,
    *,
    capability_request_script_event_id: str | None = None,
    capability_response_chat_prefix: str | None = None,
    ui_chat_chunk_prefix: str | None = None,
    session_request_chat_prefix: str | None = None,
    session_request_script_event_id: str | None = None,
    session_response_script_event_id: str | None = None,
    text_response_script_event_id: str | None = None,
    approval_allow_chat_prefix: str | None = None,
    approval_deny_chat_prefix: str | None = None,
    trusted_bridge_player_name: str | None = None,
    capability_request_schema_version: int | None = None,
    session_schema_version: int | None = None,
    text_response_framing_version: int | None = None,
    ddui_persistence_version: int | None = None,
    command_line_byte_budget: int | None = None,
    upstream_max_content_code_points: int | None = None,
    response_max_buffers: int | None = None,
    response_max_chunks_per_message: int | None = None,
    response_max_message_bytes: int | None = None,
    response_max_total_buffer_bytes: int | None = None,
    response_buffer_ttl_ms: int | None = None,
    session_response_max_command_bytes: int | None = None,
    response_chunk_delay: float = 0.15,
    response_prelude_delay: float = 0.5,
    # Deprecated aliases.  They are deliberately not stored as independent
    # values, so alias and semantic field parity cannot drift.
    bridge_request_message_id: str | None = None,
    bridge_response_prefix: str | None = None,
    ui_chat_prefix: str | None = None,
    bridge_sender: str | None = None,
    response_message_id: str | None = None,
    session_request_message_id: str | None = None,
    session_response_message_id: str | None = None,
    request_version: int | None = None,
) -> None:
    aliases = {
        "bridge_request_message_id": (
            bridge_request_message_id,
            "capability_request_script_event_id",
        ),
        "bridge_response_prefix": (bridge_response_prefix, "capability_response_chat_prefix"),
        "ui_chat_prefix": (ui_chat_prefix, "ui_chat_chunk_prefix"),
        "bridge_sender": (bridge_sender, "trusted_bridge_player_name"),
        "response_message_id": (response_message_id, "text_response_script_event_id"),
        "session_request_message_id": (
            session_request_message_id,
            "session_request_script_event_id",
        ),
        "session_response_message_id": (
            session_response_message_id,
            "session_response_script_event_id",
        ),
        "request_version": (request_version, "capability_request_schema_version"),
    }
    values: dict[str, object] = {
        "capability_request_script_event_id": capability_request_script_event_id,
        "capability_response_chat_prefix": capability_response_chat_prefix,
        "ui_chat_chunk_prefix": ui_chat_chunk_prefix,
        "session_request_chat_prefix": session_request_chat_prefix,
        "session_request_script_event_id": session_request_script_event_id,
        "session_response_script_event_id": session_response_script_event_id,
        "text_response_script_event_id": text_response_script_event_id,
        "approval_allow_chat_prefix": approval_allow_chat_prefix,
        "approval_deny_chat_prefix": approval_deny_chat_prefix,
        "trusted_bridge_player_name": trusted_bridge_player_name,
        "capability_request_schema_version": capability_request_schema_version,
        "session_schema_version": session_schema_version,
        "text_response_framing_version": text_response_framing_version,
        "ddui_persistence_version": ddui_persistence_version,
        "command_line_byte_budget": command_line_byte_budget,
        "upstream_max_content_code_points": upstream_max_content_code_points,
        "response_max_buffers": response_max_buffers,
        "response_max_chunks_per_message": response_max_chunks_per_message,
        "response_max_message_bytes": response_max_message_bytes,
        "response_max_total_buffer_bytes": response_max_total_buffer_bytes,
        "response_buffer_ttl_ms": response_buffer_ttl_ms,
        "session_response_max_command_bytes": session_response_max_command_bytes,
    }
    request_version_alias_used = request_version is not None
    for alias_name, (alias_value, semantic_name) in aliases.items():
        if alias_value is None:
            continue
        warnings.warn(
            f"McbewsV1Profile.{alias_name} is deprecated; use {semantic_name}",
            DeprecationWarning,
            stacklevel=2,
        )
        current = values[semantic_name]
        if current is not None and current != alias_value:
            raise ConfigurationError(
                f"{alias_name} conflicts with {semantic_name}; provide one name only"
            )
        values[semantic_name] = alias_value

    provided_values = {name: value for name, value in values.items() if value is not None}

    defaults: dict[str, object] = {
        "capability_request_script_event_id": _wire("capability_request_script_event_id"),
        "capability_response_chat_prefix": _wire("capability_response_chat_prefix"),
        "ui_chat_chunk_prefix": _wire("ui_chat_chunk_prefix"),
        "session_request_chat_prefix": _wire("session_request_chat_prefix"),
        "session_request_script_event_id": _wire("session_request_script_event_id"),
        "session_response_script_event_id": _wire("session_response_script_event_id"),
        "text_response_script_event_id": _wire("text_response_script_event_id"),
        "approval_allow_chat_prefix": _wire("approval_allow_chat_prefix"),
        "approval_deny_chat_prefix": _wire("approval_deny_chat_prefix"),
        "trusted_bridge_player_name": _wire("trusted_bridge_player_name"),
        "capability_request_schema_version": _version("capability_request_schema"),
        "session_schema_version": _version("session_schema"),
        "text_response_framing_version": _version("text_response_framing"),
        "ddui_persistence_version": _version("ddui_persistence"),
        "command_line_byte_budget": MCBEWS_V1_MANIFEST["limits"]["command_line_byte_budget"],
        "upstream_max_content_code_points": MCBEWS_V1_MANIFEST["limits"][
            "upstream_max_content_code_points"
        ],
        "response_max_buffers": MCBEWS_V1_MANIFEST["limits"]["response_max_buffers"],
        "response_max_chunks_per_message": MCBEWS_V1_MANIFEST["limits"][
            "response_max_chunks_per_message"
        ],
        "response_max_message_bytes": MCBEWS_V1_MANIFEST["limits"][
            "response_max_message_bytes"
        ],
        "response_max_total_buffer_bytes": MCBEWS_V1_MANIFEST["limits"][
            "response_max_total_buffer_bytes"
        ],
        "response_buffer_ttl_ms": MCBEWS_V1_MANIFEST["limits"]["response_buffer_ttl_ms"],
        "session_response_max_command_bytes": MCBEWS_V1_MANIFEST["limits"][
            "session_response_max_command_bytes"
        ],
    }
    for name, default in defaults.items():
        if values[name] is None:
            values[name] = default

    expected_capability_schema = defaults["capability_request_schema_version"]
    if values["capability_request_schema_version"] != expected_capability_schema:
        field_name = (
            "request_version"
            if request_version_alias_used
            else "capability_request_schema_version"
        )
        raise ConfigurationError(f"mcbews {field_name} must be {expected_capability_schema}")

    fixed_fields = (
        *defaults.keys(),
    )
    lowerable_limits = {"command_line_byte_budget", "session_response_max_command_bytes"}
    for name in fixed_fields:
        expected = cast(int, defaults[name])
        provided = provided_values.get(name, expected)
        if name in lowerable_limits:
            if type(provided) is not int or provided <= 0 or provided > expected:
                raise ConfigurationError(
                    f"mcbews {name} may only lower the manifest ceiling ({expected})"
                )
            values[name] = provided
            continue
        if provided != expected:
            raise ConfigurationError(f"mcbews {name} is fixed by the MCBEWS/1 manifest")
        values[name] = expected

    for name in (
        "upstream_max_content_code_points",
        "response_max_buffers",
        "response_max_chunks_per_message",
        "response_max_message_bytes",
        "response_max_total_buffer_bytes",
        "response_buffer_ttl_ms",
    ):
        value = values[name]
        if type(value) is not int or value <= 0:
            raise ConfigurationError(f"mcbews {name} must be a positive integer")
    for name in (
        "capability_request_script_event_id",
        "capability_response_chat_prefix",
        "ui_chat_chunk_prefix",
        "session_request_chat_prefix",
        "session_request_script_event_id",
        "session_response_script_event_id",
        "text_response_script_event_id",
        "approval_allow_chat_prefix",
        "approval_deny_chat_prefix",
        "trusted_bridge_player_name",
    ):
        value = values[name]
        if not isinstance(value, str) or not value.strip():
            raise ConfigurationError(f"mcbews {name} must be a non-empty string")
    _require_finite_non_negative_real(response_chunk_delay, "response_chunk_delay")
    _require_finite_non_negative_real(response_prelude_delay, "response_prelude_delay")
    protocol_line = MCBEWS_V1_MANIFEST.get("protocol_line")
    if not isinstance(protocol_line, str) or not protocol_line.strip():
        raise RuntimeError("MCBEWS/1 manifest protocol_line is invalid")
    object.__setattr__(self, "protocol_line", protocol_line)
    for name, value in values.items():
        object.__setattr__(self, name, value)
    object.__setattr__(self, "response_chunk_delay", response_chunk_delay)
    object.__setattr__(self, "response_prelude_delay", response_prelude_delay)

classify_tool_player_message 

classify_tool_player_message(sender: str, message: str, profile: McbewsV1Profile = MCBEWS_V1) -> ToolPlayerMessage | None

Authenticate and classify one ToolPlayer chat message.

None means either that the sender is not the trusted ToolPlayer or the message is ordinary player chat. Malformed reserved messages raise a typed :class:ProtocolError; callers can log/drop them without passing them to a business hook.

Source code in src/mcbe_ws_sdk/profiles/mcbews_v1/classifier.py 
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
def classify_tool_player_message(
    sender: str,
    message: str,
    profile: McbewsV1Profile = MCBEWS_V1,
) -> ToolPlayerMessage | None:
    """Authenticate and classify one ToolPlayer chat message.

    ``None`` means either that the sender is not the trusted ToolPlayer or the
    message is ordinary player chat.  Malformed reserved messages raise a typed
    :class:`ProtocolError`; callers can log/drop them without passing them to a
    business hook.
    """

    if sender != profile.trusted_bridge_player_name:
        return None
    if message.startswith(f"{profile.capability_response_chat_prefix}|"):
        return ToolPlayerMessage("bridge", sender, message)
    if message.startswith(f"{profile.ui_chat_chunk_prefix}|"):
        return ToolPlayerMessage("ui_chat", sender, message)
    if message.startswith(f"{profile.session_request_chat_prefix}|"):
        request = decode_session_request_chat(message, profile=profile)
        return ToolPlayerMessage("session", sender, message, session_request=request)
    if message.startswith(f"{profile.approval_allow_chat_prefix}|") or message.startswith(
        f"{profile.approval_deny_chat_prefix}|"
    ):
        decision = decode_approval_decision(message, profile=profile)
        return ToolPlayerMessage("approval", sender, message, approval_decision=decision)
    return None

is_tool_player_channel_message 

is_tool_player_channel_message(message: str, profile: McbewsV1Profile = MCBEWS_V1) -> bool

Return whether a message uses a reserved MCBEWS/1 control prefix.

Source code in src/mcbe_ws_sdk/profiles/mcbews_v1/classifier.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
def is_tool_player_channel_message(
    message: str,
    profile: McbewsV1Profile = MCBEWS_V1,
) -> bool:
    """Return whether a message uses a reserved MCBEWS/1 control prefix."""

    return any(
        message.startswith(prefix)
        for prefix in (
            f"{profile.capability_response_chat_prefix}|",
            f"{profile.ui_chat_chunk_prefix}|",
            f"{profile.session_request_chat_prefix}|",
            f"{profile.approval_allow_chat_prefix}|",
            f"{profile.approval_deny_chat_prefix}|",
        )
    )

decode_approval_decision 

decode_approval_decision(message: str, profile: McbewsV1Profile = MCBEWS_V1) -> ApprovalDecision

Decode a typed or one-cycle legacy ToolPlayer approval decision.

Source code in src/mcbe_ws_sdk/profiles/mcbews_v1/codec.py 
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
def decode_approval_decision(
    message: str,
    profile: McbewsV1Profile = MCBEWS_V1,
) -> ApprovalDecision:
    """Decode a typed or one-cycle legacy ToolPlayer approval decision."""

    prefixes = (
        (f"{profile.approval_allow_chat_prefix}|", "approve"),
        (f"{profile.approval_deny_chat_prefix}|", "deny"),
    )
    for prefix, decision in prefixes:
        if not message.startswith(prefix):
            continue
        payload = message[len(prefix) :]
        try:
            value = json.loads(payload)
        except JSONDecodeError:
            value = payload
        return _decode_approval_payload(value, decision=decision)  # type: ignore[arg-type]
    raise ProtocolError("invalid approval decision prefix")

decode_session_request_chat 

decode_session_request_chat(message: str, profile: McbewsV1Profile = MCBEWS_V1) -> SessionRequest

Decode MCBEWS|SESSION|<json> into a validated DTO.

Source code in src/mcbe_ws_sdk/profiles/mcbews_v1/codec.py 
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
def decode_session_request_chat(
    message: str,
    profile: McbewsV1Profile = MCBEWS_V1,
) -> SessionRequest:
    """Decode ``MCBEWS|SESSION|<json>`` into a validated DTO."""

    prefix = f"{profile.session_request_chat_prefix}|"
    if not message.startswith(prefix):
        raise ProtocolError("invalid session request prefix")
    body = message[len(prefix) :]
    try:
        value = json.loads(body)
    except JSONDecodeError as exc:
        raise ProtocolError("invalid session request JSON") from exc
    if not isinstance(value, dict):
        raise ProtocolError("session request must be a JSON object")
    return _validate_model(SessionRequest, value)

decode_session_response_message 

decode_session_response_message(message: str, profile: McbewsV1Profile = MCBEWS_V1) -> SessionResponse

Decode an Addon mcbews:session_resp event body.

Source code in src/mcbe_ws_sdk/profiles/mcbews_v1/codec.py 
314
315
316
317
318
319
320
321
322
323
324
325
326
def decode_session_response_message(
    message: str,
    profile: McbewsV1Profile = MCBEWS_V1,
) -> SessionResponse:
    """Decode an Addon ``mcbews:session_resp`` event body."""

    try:
        value = json.loads(message)
    except JSONDecodeError as exc:
        raise ProtocolError("invalid session response JSON") from exc
    if not isinstance(value, dict):
        raise ProtocolError("session response must be a JSON object")
    return _validate_model(SessionResponse, value)

decode_text_response_frame 

decode_text_response_frame(value: str | dict[str, Any]) -> TextResponseChunk

Decode and validate one text response frame, rejecting unknown roles.

Source code in src/mcbe_ws_sdk/profiles/mcbews_v1/codec.py 
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
def decode_text_response_frame(value: str | dict[str, Any]) -> TextResponseChunk:
    """Decode and validate one text response frame, rejecting unknown roles."""

    if isinstance(value, str):
        try:
            value = json.loads(value)
        except JSONDecodeError as exc:
            raise ProtocolError("invalid text response JSON") from exc
    if not isinstance(value, dict):
        raise ProtocolError("text response frame must be a JSON object")
    if value.get("r") not in {"user", "assistant", "approval"}:
        raise ProtocolError(f"UNKNOWN_TEXT_ROLE: {value.get('r')!r}")
    try:
        return TextResponseChunk.model_validate(value)
    except ValidationError as exc:
        message = str(exc)
        if "role" in message:
            message = f"UNKNOWN_TEXT_ROLE: {message}"
        raise ProtocolError(message) from exc

encode_approval_decision 

encode_approval_decision(decision: ApprovalDecision, profile: McbewsV1Profile = MCBEWS_V1) -> str

Encode a typed approval decision for the trusted ToolPlayer channel.

Source code in src/mcbe_ws_sdk/profiles/mcbews_v1/codec.py 
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
def encode_approval_decision(
    decision: ApprovalDecision,
    profile: McbewsV1Profile = MCBEWS_V1,
) -> str:
    """Encode a typed approval decision for the trusted ToolPlayer channel."""

    prefix = (
        profile.approval_allow_chat_prefix
        if decision.approved
        else profile.approval_deny_chat_prefix
    )
    if decision.legacy:
        return f"{prefix}|{decision.approval_id}"
    body = decision.model_dump(
        by_alias=True,
        exclude_none=True,
        exclude={"legacy", "decision"},
    )
    return f"{prefix}|{json.dumps(body, ensure_ascii=False, separators=(',', ':'))}"

encode_session_request_chat 

encode_session_request_chat(request: SessionRequest, profile: McbewsV1Profile = MCBEWS_V1) -> str

Encode the atomic session request chat envelope.

Source code in src/mcbe_ws_sdk/profiles/mcbews_v1/codec.py 
214
215
216
217
218
219
220
221
222
223
224
225
def encode_session_request_chat(
    request: SessionRequest,
    profile: McbewsV1Profile = MCBEWS_V1,
) -> str:
    """Encode the atomic session request chat envelope."""

    body = json.dumps(
        request.model_dump(by_alias=True, exclude_none=True),
        ensure_ascii=False,
        separators=(",", ":"),
    )
    return f"{profile.session_request_chat_prefix}|{body}"

encode_session_response_command 

encode_session_response_command(response: SessionResponse, flow: FlowControlMiddleware, profile: McbewsV1Profile = MCBEWS_V1) -> str

Encode one atomic mcbews:session_resp commandRequest payload.

A large successful result is replaced by a correlated structured error. No generic chunker is called, so a receiver never sees a fragment that looks like a complete JSON response.

Source code in src/mcbe_ws_sdk/profiles/mcbews_v1/codec.py 
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
def encode_session_response_command(
    response: SessionResponse,
    flow: FlowControlMiddleware,
    profile: McbewsV1Profile = MCBEWS_V1,
) -> str:
    """Encode one atomic ``mcbews:session_resp`` commandRequest payload.

    A large successful result is replaced by a correlated structured error.  No
    generic chunker is called, so a receiver never sees a fragment that looks
    like a complete JSON response.
    """

    command = (
        f"scriptevent {profile.session_response_script_event_id} "
        f"{encode_session_response_json(response)}"
    )
    def atomic_payload(raw_command: str) -> str:
        if len(raw_command.encode("utf-8")) > profile.session_response_max_command_bytes:
            raise FrameTooLargeError(
                "session response command exceeds the MCBEWS/1 atomic command budget"
            )
        return flow.chunk_raw_command(raw_command)[0]

    try:
        return atomic_payload(command)
    except FrameTooLargeError:
        fallback = _session_response_error(
            response,
            request_id=response.request_id,
            action=response.action,
        )
        fallback_command = (
            f"scriptevent {profile.session_response_script_event_id} "
            f"{encode_session_response_json(fallback)}"
        )
        try:
            return atomic_payload(fallback_command)
        except FrameTooLargeError as exc:
            raise FrameTooLargeError(
                "SESSION_RESPONSE_TOO_LARGE error cannot fit the configured atomic frame budget"
            ) from exc

encode_session_response_json 

encode_session_response_json(response: SessionResponse) -> str

Serialize a session response with compact wire aliases.

Source code in src/mcbe_ws_sdk/profiles/mcbews_v1/codec.py 
247
248
249
250
251
252
253
254
def encode_session_response_json(response: SessionResponse) -> str:
    """Serialize a session response with compact wire aliases."""

    return json.dumps(
        response.model_dump(by_alias=True, exclude_none=True),
        ensure_ascii=False,
        separators=(",", ":"),
    )

reassemble_text_response_chunks 

reassemble_text_response_chunks(chunks: list[TextResponseChunk]) -> TextResponseMessage

Reassemble already validated text chunks with metadata consistency checks.

Source code in src/mcbe_ws_sdk/profiles/mcbews_v1/codec.py 
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
def reassemble_text_response_chunks(chunks: list[TextResponseChunk]) -> TextResponseMessage:
    """Reassemble already validated text chunks with metadata consistency checks."""

    if not chunks:
        raise ProtocolError("text response chunks must not be empty")
    first = chunks[0]
    if any(
        chunk.response_id != first.response_id
        or chunk.total != first.total
        or chunk.player_name != first.player_name
        or chunk.role != first.role
        or chunk.conversation_id != first.conversation_id
        or chunk.title != first.title
        for chunk in chunks
    ):
        raise ProtocolError("text response metadata conflict")
    if first.total > MCBEWS_V1.response_max_chunks_per_message:
        raise ProtocolError("text response chunk count exceeds configured limit")
    by_index: dict[int, TextResponseChunk] = {}
    total_bytes = 0
    for chunk in chunks:
        previous = by_index.get(chunk.index)
        if previous is not None:
            if previous.content != chunk.content or previous.usage != chunk.usage:
                raise ProtocolError("text response duplicate metadata conflict")
            continue
        by_index[chunk.index] = chunk
        total_bytes += len(chunk.content.encode("utf-8"))
    if total_bytes > MCBEWS_V1.response_max_message_bytes:
        raise ProtocolError("text response message exceeds configured byte limit")
    expected = set(range(1, first.total + 1))
    if set(by_index) != expected:
        raise ProtocolError("text response chunks are incomplete")
    final = by_index[first.total]
    return TextResponseMessage(
        response_id=first.response_id,
        player_name=first.player_name,
        role=first.role,
        content="".join(by_index[index].content for index in range(1, first.total + 1)),
        conversation_id=first.conversation_id,
        title=first.title,
        usage=final.usage,
    )

load_manifest 

load_manifest() -> Mapping[str, Any]

Load the immutable MCBEWS/1 manifest from the installed package.

Source code in src/mcbe_ws_sdk/profiles/mcbews_v1/manifest.py 
35
36
37
38
def load_manifest() -> Mapping[str, Any]:
    """Load the immutable MCBEWS/1 manifest from the installed package."""

    return cast(Mapping[str, Any], _freeze(_load_json("manifest.json")))

load_wire_vectors 

load_wire_vectors() -> Mapping[str, Any]

Load the immutable executable MCBEWS/1 conformance vectors.

Source code in src/mcbe_ws_sdk/profiles/mcbews_v1/manifest.py 
41
42
43
44
def load_wire_vectors() -> Mapping[str, Any]:
    """Load the immutable executable MCBEWS/1 conformance vectors."""

    return cast(Mapping[str, Any], _freeze(_load_json("vectors.json")))

classifier

Trusted ToolPlayer channel classification for MCBEWS/1.

Classes:

  • ToolPlayerMessage –

    A control-channel message accepted only from the trusted ToolPlayer.

Functions:

ToolPlayerMessage dataclass 
ToolPlayerMessage(channel: ToolPlayerChannel, sender: str, raw_message: str, session_request: SessionRequest | None = None, approval_decision: ApprovalDecision | None = None)

A control-channel message accepted only from the trusted ToolPlayer.

is_tool_player_channel_message 
is_tool_player_channel_message(message: str, profile: McbewsV1Profile = MCBEWS_V1) -> bool

Return whether a message uses a reserved MCBEWS/1 control prefix.

Source code in src/mcbe_ws_sdk/profiles/mcbews_v1/classifier.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
def is_tool_player_channel_message(
    message: str,
    profile: McbewsV1Profile = MCBEWS_V1,
) -> bool:
    """Return whether a message uses a reserved MCBEWS/1 control prefix."""

    return any(
        message.startswith(prefix)
        for prefix in (
            f"{profile.capability_response_chat_prefix}|",
            f"{profile.ui_chat_chunk_prefix}|",
            f"{profile.session_request_chat_prefix}|",
            f"{profile.approval_allow_chat_prefix}|",
            f"{profile.approval_deny_chat_prefix}|",
        )
    )
classify_tool_player_message 
classify_tool_player_message(sender: str, message: str, profile: McbewsV1Profile = MCBEWS_V1) -> ToolPlayerMessage | None

Authenticate and classify one ToolPlayer chat message.

None means either that the sender is not the trusted ToolPlayer or the message is ordinary player chat. Malformed reserved messages raise a typed :class:ProtocolError; callers can log/drop them without passing them to a business hook.

Source code in src/mcbe_ws_sdk/profiles/mcbews_v1/classifier.py 
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
def classify_tool_player_message(
    sender: str,
    message: str,
    profile: McbewsV1Profile = MCBEWS_V1,
) -> ToolPlayerMessage | None:
    """Authenticate and classify one ToolPlayer chat message.

    ``None`` means either that the sender is not the trusted ToolPlayer or the
    message is ordinary player chat.  Malformed reserved messages raise a typed
    :class:`ProtocolError`; callers can log/drop them without passing them to a
    business hook.
    """

    if sender != profile.trusted_bridge_player_name:
        return None
    if message.startswith(f"{profile.capability_response_chat_prefix}|"):
        return ToolPlayerMessage("bridge", sender, message)
    if message.startswith(f"{profile.ui_chat_chunk_prefix}|"):
        return ToolPlayerMessage("ui_chat", sender, message)
    if message.startswith(f"{profile.session_request_chat_prefix}|"):
        request = decode_session_request_chat(message, profile=profile)
        return ToolPlayerMessage("session", sender, message, session_request=request)
    if message.startswith(f"{profile.approval_allow_chat_prefix}|") or message.startswith(
        f"{profile.approval_deny_chat_prefix}|"
    ):
        decision = decode_approval_decision(message, profile=profile)
        return ToolPlayerMessage("approval", sender, message, approval_decision=decision)
    return None

codec

Functions:

encode_session_request_chat 
encode_session_request_chat(request: SessionRequest, profile: McbewsV1Profile = MCBEWS_V1) -> str

Encode the atomic session request chat envelope.

Source code in src/mcbe_ws_sdk/profiles/mcbews_v1/codec.py 
214
215
216
217
218
219
220
221
222
223
224
225
def encode_session_request_chat(
    request: SessionRequest,
    profile: McbewsV1Profile = MCBEWS_V1,
) -> str:
    """Encode the atomic session request chat envelope."""

    body = json.dumps(
        request.model_dump(by_alias=True, exclude_none=True),
        ensure_ascii=False,
        separators=(",", ":"),
    )
    return f"{profile.session_request_chat_prefix}|{body}"
decode_session_request_chat 
decode_session_request_chat(message: str, profile: McbewsV1Profile = MCBEWS_V1) -> SessionRequest

Decode MCBEWS|SESSION|<json> into a validated DTO.

Source code in src/mcbe_ws_sdk/profiles/mcbews_v1/codec.py 
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
def decode_session_request_chat(
    message: str,
    profile: McbewsV1Profile = MCBEWS_V1,
) -> SessionRequest:
    """Decode ``MCBEWS|SESSION|<json>`` into a validated DTO."""

    prefix = f"{profile.session_request_chat_prefix}|"
    if not message.startswith(prefix):
        raise ProtocolError("invalid session request prefix")
    body = message[len(prefix) :]
    try:
        value = json.loads(body)
    except JSONDecodeError as exc:
        raise ProtocolError("invalid session request JSON") from exc
    if not isinstance(value, dict):
        raise ProtocolError("session request must be a JSON object")
    return _validate_model(SessionRequest, value)
encode_session_response_json 
encode_session_response_json(response: SessionResponse) -> str

Serialize a session response with compact wire aliases.

Source code in src/mcbe_ws_sdk/profiles/mcbews_v1/codec.py 
247
248
249
250
251
252
253
254
def encode_session_response_json(response: SessionResponse) -> str:
    """Serialize a session response with compact wire aliases."""

    return json.dumps(
        response.model_dump(by_alias=True, exclude_none=True),
        ensure_ascii=False,
        separators=(",", ":"),
    )
encode_session_response_command 
encode_session_response_command(response: SessionResponse, flow: FlowControlMiddleware, profile: McbewsV1Profile = MCBEWS_V1) -> str

Encode one atomic mcbews:session_resp commandRequest payload.

A large successful result is replaced by a correlated structured error. No generic chunker is called, so a receiver never sees a fragment that looks like a complete JSON response.

Source code in src/mcbe_ws_sdk/profiles/mcbews_v1/codec.py 
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
def encode_session_response_command(
    response: SessionResponse,
    flow: FlowControlMiddleware,
    profile: McbewsV1Profile = MCBEWS_V1,
) -> str:
    """Encode one atomic ``mcbews:session_resp`` commandRequest payload.

    A large successful result is replaced by a correlated structured error.  No
    generic chunker is called, so a receiver never sees a fragment that looks
    like a complete JSON response.
    """

    command = (
        f"scriptevent {profile.session_response_script_event_id} "
        f"{encode_session_response_json(response)}"
    )
    def atomic_payload(raw_command: str) -> str:
        if len(raw_command.encode("utf-8")) > profile.session_response_max_command_bytes:
            raise FrameTooLargeError(
                "session response command exceeds the MCBEWS/1 atomic command budget"
            )
        return flow.chunk_raw_command(raw_command)[0]

    try:
        return atomic_payload(command)
    except FrameTooLargeError:
        fallback = _session_response_error(
            response,
            request_id=response.request_id,
            action=response.action,
        )
        fallback_command = (
            f"scriptevent {profile.session_response_script_event_id} "
            f"{encode_session_response_json(fallback)}"
        )
        try:
            return atomic_payload(fallback_command)
        except FrameTooLargeError as exc:
            raise FrameTooLargeError(
                "SESSION_RESPONSE_TOO_LARGE error cannot fit the configured atomic frame budget"
            ) from exc
decode_session_response_message 
decode_session_response_message(message: str, profile: McbewsV1Profile = MCBEWS_V1) -> SessionResponse

Decode an Addon mcbews:session_resp event body.

Source code in src/mcbe_ws_sdk/profiles/mcbews_v1/codec.py 
314
315
316
317
318
319
320
321
322
323
324
325
326
def decode_session_response_message(
    message: str,
    profile: McbewsV1Profile = MCBEWS_V1,
) -> SessionResponse:
    """Decode an Addon ``mcbews:session_resp`` event body."""

    try:
        value = json.loads(message)
    except JSONDecodeError as exc:
        raise ProtocolError("invalid session response JSON") from exc
    if not isinstance(value, dict):
        raise ProtocolError("session response must be a JSON object")
    return _validate_model(SessionResponse, value)
decode_approval_decision 
decode_approval_decision(message: str, profile: McbewsV1Profile = MCBEWS_V1) -> ApprovalDecision

Decode a typed or one-cycle legacy ToolPlayer approval decision.

Source code in src/mcbe_ws_sdk/profiles/mcbews_v1/codec.py 
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
def decode_approval_decision(
    message: str,
    profile: McbewsV1Profile = MCBEWS_V1,
) -> ApprovalDecision:
    """Decode a typed or one-cycle legacy ToolPlayer approval decision."""

    prefixes = (
        (f"{profile.approval_allow_chat_prefix}|", "approve"),
        (f"{profile.approval_deny_chat_prefix}|", "deny"),
    )
    for prefix, decision in prefixes:
        if not message.startswith(prefix):
            continue
        payload = message[len(prefix) :]
        try:
            value = json.loads(payload)
        except JSONDecodeError:
            value = payload
        return _decode_approval_payload(value, decision=decision)  # type: ignore[arg-type]
    raise ProtocolError("invalid approval decision prefix")
decode_text_response_frame 
decode_text_response_frame(value: str | dict[str, Any]) -> TextResponseChunk

Decode and validate one text response frame, rejecting unknown roles.

Source code in src/mcbe_ws_sdk/profiles/mcbews_v1/codec.py 
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
def decode_text_response_frame(value: str | dict[str, Any]) -> TextResponseChunk:
    """Decode and validate one text response frame, rejecting unknown roles."""

    if isinstance(value, str):
        try:
            value = json.loads(value)
        except JSONDecodeError as exc:
            raise ProtocolError("invalid text response JSON") from exc
    if not isinstance(value, dict):
        raise ProtocolError("text response frame must be a JSON object")
    if value.get("r") not in {"user", "assistant", "approval"}:
        raise ProtocolError(f"UNKNOWN_TEXT_ROLE: {value.get('r')!r}")
    try:
        return TextResponseChunk.model_validate(value)
    except ValidationError as exc:
        message = str(exc)
        if "role" in message:
            message = f"UNKNOWN_TEXT_ROLE: {message}"
        raise ProtocolError(message) from exc
reassemble_text_response_chunks 
reassemble_text_response_chunks(chunks: list[TextResponseChunk]) -> TextResponseMessage

Reassemble already validated text chunks with metadata consistency checks.

Source code in src/mcbe_ws_sdk/profiles/mcbews_v1/codec.py 
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
def reassemble_text_response_chunks(chunks: list[TextResponseChunk]) -> TextResponseMessage:
    """Reassemble already validated text chunks with metadata consistency checks."""

    if not chunks:
        raise ProtocolError("text response chunks must not be empty")
    first = chunks[0]
    if any(
        chunk.response_id != first.response_id
        or chunk.total != first.total
        or chunk.player_name != first.player_name
        or chunk.role != first.role
        or chunk.conversation_id != first.conversation_id
        or chunk.title != first.title
        for chunk in chunks
    ):
        raise ProtocolError("text response metadata conflict")
    if first.total > MCBEWS_V1.response_max_chunks_per_message:
        raise ProtocolError("text response chunk count exceeds configured limit")
    by_index: dict[int, TextResponseChunk] = {}
    total_bytes = 0
    for chunk in chunks:
        previous = by_index.get(chunk.index)
        if previous is not None:
            if previous.content != chunk.content or previous.usage != chunk.usage:
                raise ProtocolError("text response duplicate metadata conflict")
            continue
        by_index[chunk.index] = chunk
        total_bytes += len(chunk.content.encode("utf-8"))
    if total_bytes > MCBEWS_V1.response_max_message_bytes:
        raise ProtocolError("text response message exceeds configured byte limit")
    expected = set(range(1, first.total + 1))
    if set(by_index) != expected:
        raise ProtocolError("text response chunks are incomplete")
    final = by_index[first.total]
    return TextResponseMessage(
        response_id=first.response_id,
        player_name=first.player_name,
        role=first.role,
        content="".join(by_index[index].content for index in range(1, first.total + 1)),
        conversation_id=first.conversation_id,
        title=first.title,
        usage=final.usage,
    )
encode_approval_decision 
encode_approval_decision(decision: ApprovalDecision, profile: McbewsV1Profile = MCBEWS_V1) -> str

Encode a typed approval decision for the trusted ToolPlayer channel.

Source code in src/mcbe_ws_sdk/profiles/mcbews_v1/codec.py 
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
def encode_approval_decision(
    decision: ApprovalDecision,
    profile: McbewsV1Profile = MCBEWS_V1,
) -> str:
    """Encode a typed approval decision for the trusted ToolPlayer channel."""

    prefix = (
        profile.approval_allow_chat_prefix
        if decision.approved
        else profile.approval_deny_chat_prefix
    )
    if decision.legacy:
        return f"{prefix}|{decision.approval_id}"
    body = decision.model_dump(
        by_alias=True,
        exclude_none=True,
        exclude={"legacy", "decision"},
    )
    return f"{prefix}|{json.dumps(body, ensure_ascii=False, separators=(',', ':'))}"

delivery

Classes:

McbewsV1Delivery 
McbewsV1Delivery(outbound: McbeOutboundDelivery, *, profile: McbewsV1Profile = MCBEWS_V1, sleeper: Sleeper = asyncio.sleep)

Methods:

Source code in src/mcbe_ws_sdk/profiles/mcbews_v1/delivery.py 
18
19
20
21
22
23
24
25
26
27
def __init__(
    self,
    outbound: McbeOutboundDelivery,
    *,
    profile: McbewsV1Profile = MCBEWS_V1,
    sleeper: Sleeper = asyncio.sleep,
) -> None:
    self._outbound = outbound
    self._profile = profile
    self._sleep = sleeper
send_session_response async 
send_session_response(response: SessionResponse) -> None

Send one atomic correlated session response.

Session responses deliberately bypass generic scriptevent chunking. A result that does not fit is encoded as one SESSION_RESPONSE_TOO_LARGE error by :func:encode_session_response_command.

Source code in src/mcbe_ws_sdk/profiles/mcbews_v1/delivery.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
async def send_session_response(self, response: SessionResponse) -> None:
    """Send one atomic correlated session response.

    Session responses deliberately bypass generic scriptevent chunking. A
    result that does not fit is encoded as one ``SESSION_RESPONSE_TOO_LARGE``
    error by :func:`encode_session_response_command`.
    """

    payload = encode_session_response_command(
        response,
        self._outbound.flow,
        profile=self._profile,
    )
    await self._outbound.send_payload(payload, "session_resp")

manifest

Canonical MCBEWS/1 protocol assets.

The JSON resources in this module are intentionally small and executable. The Python codec, the reference Addon and downstream conformance checks all project their constants and vectors from the same shape instead of treating a hand copied list of strings as the protocol authority.

Functions:

  • load_manifest –

    Load the immutable MCBEWS/1 manifest from the installed package.

  • load_wire_vectors –

    Load the immutable executable MCBEWS/1 conformance vectors.

load_manifest 
load_manifest() -> Mapping[str, Any]

Load the immutable MCBEWS/1 manifest from the installed package.

Source code in src/mcbe_ws_sdk/profiles/mcbews_v1/manifest.py 
35
36
37
38
def load_manifest() -> Mapping[str, Any]:
    """Load the immutable MCBEWS/1 manifest from the installed package."""

    return cast(Mapping[str, Any], _freeze(_load_json("manifest.json")))
load_wire_vectors 
load_wire_vectors() -> Mapping[str, Any]

Load the immutable executable MCBEWS/1 conformance vectors.

Source code in src/mcbe_ws_sdk/profiles/mcbews_v1/manifest.py 
41
42
43
44
def load_wire_vectors() -> Mapping[str, Any]:
    """Load the immutable executable MCBEWS/1 conformance vectors."""

    return cast(Mapping[str, Any], _freeze(_load_json("vectors.json")))

models

Typed MCBEWS/1 wire DTOs.

The models intentionally keep the compact wire aliases (v, cid, sid and u) at the boundary while exposing semantic Python names to Host adapters.

Classes:

  • AddonBridgeRequest –

    Capability request sent to the reference Addon.

  • UiChatMessage –

    A complete UI chat message with its originating conversation.

  • SessionRequest –

    Validated session operation request from the trusted ToolPlayer.

  • SessionError –

    Stable, bounded session error object.

  • SessionResponse –

    Atomic session response correlated to one :class:SessionRequest.

  • ApprovalDecision –

    Typed approval decision carried by the ToolPlayer chat channel.

  • TokenUsage –

    Compact token usage included only in a final text response frame.

  • TextResponseChunk –

    One validated mcbews:text_resp frame.

  • TextResponseMessage –

    A complete, ordered text response after bounded reassembly.

AddonBridgeRequest

Bases: BaseModel

Capability request sent to the reference Addon.

UiChatMessage

Bases: BaseModel

A complete UI chat message with its originating conversation.

SessionRequest

Bases: BaseModel

Validated session operation request from the trusted ToolPlayer.

SessionError

Bases: BaseModel

Stable, bounded session error object.

SessionResponse

Bases: BaseModel

Atomic session response correlated to one :class:SessionRequest.

ApprovalDecision

Bases: BaseModel

Typed approval decision carried by the ToolPlayer chat channel.

legacy=True is only used by the decoder for an id-only decision. Such a decision deliberately has no owner and must be resolved by a Host pending approval store; it is never treated as coming from the transport sender.

Attributes:

  • approved (bool) –

    Whether this decision approves the pending operation.

approved property 
approved: bool

Whether this decision approves the pending operation.

TokenUsage

Bases: BaseModel

Compact token usage included only in a final text response frame.

TextResponseChunk

Bases: BaseModel

One validated mcbews:text_resp frame.

TextResponseMessage

Bases: BaseModel

A complete, ordered text response after bounded reassembly.

protocol_error_message 
protocol_error_message(error: Exception) -> ProtocolError

Convert a Pydantic validation error to the SDK boundary error type.

Source code in src/mcbe_ws_sdk/profiles/mcbews_v1/models.py 
285
286
287
288
def protocol_error_message(error: Exception) -> ProtocolError:
    """Convert a Pydantic validation error to the SDK boundary error type."""

    return ProtocolError(str(error))

profile

Classes:

McbewsV1Profile dataclass 
McbewsV1Profile(*, capability_request_script_event_id: str | None = None, capability_response_chat_prefix: str | None = None, ui_chat_chunk_prefix: str | None = None, session_request_chat_prefix: str | None = None, session_request_script_event_id: str | None = None, session_response_script_event_id: str | None = None, text_response_script_event_id: str | None = None, approval_allow_chat_prefix: str | None = None, approval_deny_chat_prefix: str | None = None, trusted_bridge_player_name: str | None = None, capability_request_schema_version: int | None = None, session_schema_version: int | None = None, text_response_framing_version: int | None = None, ddui_persistence_version: int | None = None, command_line_byte_budget: int | None = None, upstream_max_content_code_points: int | None = None, response_max_buffers: int | None = None, response_max_chunks_per_message: int | None = None, response_max_message_bytes: int | None = None, response_max_total_buffer_bytes: int | None = None, response_buffer_ttl_ms: int | None = None, session_response_max_command_bytes: int | None = None, response_chunk_delay: float = 0.15, response_prelude_delay: float = 0.5, bridge_request_message_id: str | None = None, bridge_response_prefix: str | None = None, ui_chat_prefix: str | None = None, bridge_sender: str | None = None, response_message_id: str | None = None, session_request_message_id: str | None = None, session_response_message_id: str | None = None, request_version: int | None = None)

Concrete MCBEWS/1 wire profile.

The semantic field names are the protocol authority. Wire identifiers, schema versions and safety bounds come from the installed manifest and cannot be replaced by a runtime profile seam. The empirical command budget may be lowered for a deployment and response delays remain operational knobs. Historical names are accepted for one migration cycle only when they still select the manifest value.

Source code in src/mcbe_ws_sdk/profiles/mcbews_v1/profile.py 
 70
 71
 72
 73
 74
 75
 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
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
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
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
def __init__(
    self,
    *,
    capability_request_script_event_id: str | None = None,
    capability_response_chat_prefix: str | None = None,
    ui_chat_chunk_prefix: str | None = None,
    session_request_chat_prefix: str | None = None,
    session_request_script_event_id: str | None = None,
    session_response_script_event_id: str | None = None,
    text_response_script_event_id: str | None = None,
    approval_allow_chat_prefix: str | None = None,
    approval_deny_chat_prefix: str | None = None,
    trusted_bridge_player_name: str | None = None,
    capability_request_schema_version: int | None = None,
    session_schema_version: int | None = None,
    text_response_framing_version: int | None = None,
    ddui_persistence_version: int | None = None,
    command_line_byte_budget: int | None = None,
    upstream_max_content_code_points: int | None = None,
    response_max_buffers: int | None = None,
    response_max_chunks_per_message: int | None = None,
    response_max_message_bytes: int | None = None,
    response_max_total_buffer_bytes: int | None = None,
    response_buffer_ttl_ms: int | None = None,
    session_response_max_command_bytes: int | None = None,
    response_chunk_delay: float = 0.15,
    response_prelude_delay: float = 0.5,
    # Deprecated aliases.  They are deliberately not stored as independent
    # values, so alias and semantic field parity cannot drift.
    bridge_request_message_id: str | None = None,
    bridge_response_prefix: str | None = None,
    ui_chat_prefix: str | None = None,
    bridge_sender: str | None = None,
    response_message_id: str | None = None,
    session_request_message_id: str | None = None,
    session_response_message_id: str | None = None,
    request_version: int | None = None,
) -> None:
    aliases = {
        "bridge_request_message_id": (
            bridge_request_message_id,
            "capability_request_script_event_id",
        ),
        "bridge_response_prefix": (bridge_response_prefix, "capability_response_chat_prefix"),
        "ui_chat_prefix": (ui_chat_prefix, "ui_chat_chunk_prefix"),
        "bridge_sender": (bridge_sender, "trusted_bridge_player_name"),
        "response_message_id": (response_message_id, "text_response_script_event_id"),
        "session_request_message_id": (
            session_request_message_id,
            "session_request_script_event_id",
        ),
        "session_response_message_id": (
            session_response_message_id,
            "session_response_script_event_id",
        ),
        "request_version": (request_version, "capability_request_schema_version"),
    }
    values: dict[str, object] = {
        "capability_request_script_event_id": capability_request_script_event_id,
        "capability_response_chat_prefix": capability_response_chat_prefix,
        "ui_chat_chunk_prefix": ui_chat_chunk_prefix,
        "session_request_chat_prefix": session_request_chat_prefix,
        "session_request_script_event_id": session_request_script_event_id,
        "session_response_script_event_id": session_response_script_event_id,
        "text_response_script_event_id": text_response_script_event_id,
        "approval_allow_chat_prefix": approval_allow_chat_prefix,
        "approval_deny_chat_prefix": approval_deny_chat_prefix,
        "trusted_bridge_player_name": trusted_bridge_player_name,
        "capability_request_schema_version": capability_request_schema_version,
        "session_schema_version": session_schema_version,
        "text_response_framing_version": text_response_framing_version,
        "ddui_persistence_version": ddui_persistence_version,
        "command_line_byte_budget": command_line_byte_budget,
        "upstream_max_content_code_points": upstream_max_content_code_points,
        "response_max_buffers": response_max_buffers,
        "response_max_chunks_per_message": response_max_chunks_per_message,
        "response_max_message_bytes": response_max_message_bytes,
        "response_max_total_buffer_bytes": response_max_total_buffer_bytes,
        "response_buffer_ttl_ms": response_buffer_ttl_ms,
        "session_response_max_command_bytes": session_response_max_command_bytes,
    }
    request_version_alias_used = request_version is not None
    for alias_name, (alias_value, semantic_name) in aliases.items():
        if alias_value is None:
            continue
        warnings.warn(
            f"McbewsV1Profile.{alias_name} is deprecated; use {semantic_name}",
            DeprecationWarning,
            stacklevel=2,
        )
        current = values[semantic_name]
        if current is not None and current != alias_value:
            raise ConfigurationError(
                f"{alias_name} conflicts with {semantic_name}; provide one name only"
            )
        values[semantic_name] = alias_value

    provided_values = {name: value for name, value in values.items() if value is not None}

    defaults: dict[str, object] = {
        "capability_request_script_event_id": _wire("capability_request_script_event_id"),
        "capability_response_chat_prefix": _wire("capability_response_chat_prefix"),
        "ui_chat_chunk_prefix": _wire("ui_chat_chunk_prefix"),
        "session_request_chat_prefix": _wire("session_request_chat_prefix"),
        "session_request_script_event_id": _wire("session_request_script_event_id"),
        "session_response_script_event_id": _wire("session_response_script_event_id"),
        "text_response_script_event_id": _wire("text_response_script_event_id"),
        "approval_allow_chat_prefix": _wire("approval_allow_chat_prefix"),
        "approval_deny_chat_prefix": _wire("approval_deny_chat_prefix"),
        "trusted_bridge_player_name": _wire("trusted_bridge_player_name"),
        "capability_request_schema_version": _version("capability_request_schema"),
        "session_schema_version": _version("session_schema"),
        "text_response_framing_version": _version("text_response_framing"),
        "ddui_persistence_version": _version("ddui_persistence"),
        "command_line_byte_budget": MCBEWS_V1_MANIFEST["limits"]["command_line_byte_budget"],
        "upstream_max_content_code_points": MCBEWS_V1_MANIFEST["limits"][
            "upstream_max_content_code_points"
        ],
        "response_max_buffers": MCBEWS_V1_MANIFEST["limits"]["response_max_buffers"],
        "response_max_chunks_per_message": MCBEWS_V1_MANIFEST["limits"][
            "response_max_chunks_per_message"
        ],
        "response_max_message_bytes": MCBEWS_V1_MANIFEST["limits"][
            "response_max_message_bytes"
        ],
        "response_max_total_buffer_bytes": MCBEWS_V1_MANIFEST["limits"][
            "response_max_total_buffer_bytes"
        ],
        "response_buffer_ttl_ms": MCBEWS_V1_MANIFEST["limits"]["response_buffer_ttl_ms"],
        "session_response_max_command_bytes": MCBEWS_V1_MANIFEST["limits"][
            "session_response_max_command_bytes"
        ],
    }
    for name, default in defaults.items():
        if values[name] is None:
            values[name] = default

    expected_capability_schema = defaults["capability_request_schema_version"]
    if values["capability_request_schema_version"] != expected_capability_schema:
        field_name = (
            "request_version"
            if request_version_alias_used
            else "capability_request_schema_version"
        )
        raise ConfigurationError(f"mcbews {field_name} must be {expected_capability_schema}")

    fixed_fields = (
        *defaults.keys(),
    )
    lowerable_limits = {"command_line_byte_budget", "session_response_max_command_bytes"}
    for name in fixed_fields:
        expected = cast(int, defaults[name])
        provided = provided_values.get(name, expected)
        if name in lowerable_limits:
            if type(provided) is not int or provided <= 0 or provided > expected:
                raise ConfigurationError(
                    f"mcbews {name} may only lower the manifest ceiling ({expected})"
                )
            values[name] = provided
            continue
        if provided != expected:
            raise ConfigurationError(f"mcbews {name} is fixed by the MCBEWS/1 manifest")
        values[name] = expected

    for name in (
        "upstream_max_content_code_points",
        "response_max_buffers",
        "response_max_chunks_per_message",
        "response_max_message_bytes",
        "response_max_total_buffer_bytes",
        "response_buffer_ttl_ms",
    ):
        value = values[name]
        if type(value) is not int or value <= 0:
            raise ConfigurationError(f"mcbews {name} must be a positive integer")
    for name in (
        "capability_request_script_event_id",
        "capability_response_chat_prefix",
        "ui_chat_chunk_prefix",
        "session_request_chat_prefix",
        "session_request_script_event_id",
        "session_response_script_event_id",
        "text_response_script_event_id",
        "approval_allow_chat_prefix",
        "approval_deny_chat_prefix",
        "trusted_bridge_player_name",
    ):
        value = values[name]
        if not isinstance(value, str) or not value.strip():
            raise ConfigurationError(f"mcbews {name} must be a non-empty string")
    _require_finite_non_negative_real(response_chunk_delay, "response_chunk_delay")
    _require_finite_non_negative_real(response_prelude_delay, "response_prelude_delay")
    protocol_line = MCBEWS_V1_MANIFEST.get("protocol_line")
    if not isinstance(protocol_line, str) or not protocol_line.strip():
        raise RuntimeError("MCBEWS/1 manifest protocol_line is invalid")
    object.__setattr__(self, "protocol_line", protocol_line)
    for name, value in values.items():
        object.__setattr__(self, name, value)
    object.__setattr__(self, "response_chunk_delay", response_chunk_delay)
    object.__setattr__(self, "response_prelude_delay", response_prelude_delay)

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.