packets API

完整接口目录 · 协议包用法 · 工具包下载

模块说明

Task-owned packet listeners. Payloads remain tagged custom bytes or JSON bytes.

Packets

class Packets

Packets.__init__

def __init__(self, context)

Packets.typed

@property
def typed(self)

SDK 说明：

Versioned protocol models and connection-fenced generic packet APIs.

Packets.catalog

async def catalog(self)

SDK 说明：

Static version-scoped metadata, unchanged; unknown is not permission.

Encoding registration and conditional transport never prove server
acceptance. Legacy observable describes the observation adapter only.

Packets.prepare_raw

def prepare_raw(self, frame, *, connection_id, idempotency_key, deadline)

SDK 说明：

Prepare an immutable raw send intent; persist it before send_raw.

Packets.send_raw

async def send_raw(self, intent, *, timeout=10)

SDK 说明：

Submit one prepared raw intent. Unknown results must not be replayed.

Packets.describe

async def describe(self, packet_id)

SDK 说明：

Registered forms and audit metadata; never reads the world.

Requires Host API minor 3. New metadata is additive: older Hosts may
omit it. Missing/unknown direction or execution must remain unknown.

Packets.preview

async def preview(
    self,
    packet_id,
    packet,
    *,
    format = 'original_json',
    packet_type = None,
    context = 'offline',
)

SDK 说明：

Encode a diagnostic body without sending or submitting an operation.

Requires Host API minor 3 and packet_preview_ids. HeldItemRef requires
context='inventory_snapshot' plus self-inventory observe permission.
OK means encoding succeeded, never server acceptance. Body uses shield
ID zero and omits the header: submit the original intent for live sends.

Packets.encode

async def encode(
    self,
    packet_id,
    packet,
    *,
    packet_type = None,
    format = 'lossless',
    context = 'offline',
)

SDK 说明：

Encode a lossless object to a headerless offline body, without sending.

Requires Host API minor 3 and packet_preview_ids. The Host uses the same
encoder as preview and caps the body at 128 KiB. Returns diagnostics
unchanged; OK only means encoding succeeded.

Packets.decode

async def decode(
    self,
    packet_id,
    body,
    *,
    packet_type = None,
    format = 'lossless',
    context = 'offline',
)

SDK 说明：

Decode complete headerless bytes into a lossless object, without sending.

Accepts bytes, bytearray or memoryview, up to 128 KiB. Requires Host API
minor 3 and packet_preview_ids. Uses offline shield ID zero; live frames
must not be passed here. Returns the Host's diagnostic result unchanged.

Packets.open

async def open(self, packet_ids, *, text_profile=None)

Packets.listen

async def listen(self, packet_ids, callback, *, on_snapshot=None, text_profile=None)

SDK 说明：

Open before returning; callback backpressure is serial, never task-per-packet.

Returns a Worker-owned asyncio Task. cancel() then await it to release.
Overflow/revocation fails the listener; never silently reconnect or replay.

Packets.context

self.context

公开实例属性，在 __init__ 中设置。