huawei-cloud-cloudrobo-inference-adapter

Guides adapting a VLA model's inference server interface to the CloudRobo standard interface/protocol. Use this skill when the user wants to create a compatible inference interface protocol for a model, or make a model compatible with the CloudRobo client library. Trigger conditions: "adapt inference interface", "inference service compatibility", "flattened interface", or any request to make a model server compatible with an external inference client library.

Inference Interface Adapter

Overview

This skill guides converting a VLA model's inference interface into an inference server that uses a flattened linear format, adapting to the CloudRobo client library.

Core idea: the model's own inference interface format differs from the CloudRobo standard communication format. Instead of modifying the model's inference logic, add a thin adapter layer that converts between the flattened linear format and the model's internal format.

Fixed Protocol Contract

Item	Convention
Transport	WSS (WebSocket over TLS)
Certificate	Self-signed certificate generated at runtime (cryptography lib, RSA 3072)
Serialization	msgpack (+ numpy extension)
Observation format	Flat dot-notation dict
Observation keys	observation.images.{key} (key derived from the model config, not hardcoded), observation.state (fixed field), task (fixed field)
Action response	{"action": (T, D) float32}, fixed field
Handshake	First frame after connect sends the metadata contract
Error	Send traceback.format_exc() string, close with INTERNAL_ERROR

Observation request example:

python

{
    "observation.state": np.ndarray,              # (D,) float32
    "observation.images.front": np.ndarray,   # (H, W, 3) uint8
    "observation.images.wrist_left": np.ndarray,    # (H, W, 3) uint8
    "task": str,
}

Action response example:

python

{
    "action": np.ndarray,                         # (T, D) float32
    "server_timing": {"infer_ms": ..., "prev_total_ms": ...},
}

Prerequisites

1. The model's existing inference interface — a predict/infer method, model metadata (action_dim, state_dim, image_keys, action_chunk_size), and normalization access (how to get/set denormalization parameters).
2. Training outputs (for the entry point) — the final weights from the standard layout plus dataset_statistics.json (required for action denormalization).
3. Python packages — numpy, msgpack (serialization with the __ndarray__/__npgeneric__ numpy extension), websockets (server + client), and cryptography (self-signed certificates). Pillow is optional (loading images from disk in the test client).

Built-in Scripts

This skill ships two reusable scripts in scripts/:

• test_client.py — a generic WSS test client that auto-discovers the server's I/O contract from the handshake metadata. Supports random images or loading from disk, single/multiple inferences, and performance benchmarking. Used to end-to-end verify any adapter server. Run python scripts/test_client.py after starting the server.

• verify_adapter.py — a unit-testing tool for the adapter layer. Tests only through adapter.infer(obs) (no model API assumptions), covering observation conversion, image-key validation, state-dimension validation, and the metadata property — all without running a server or loading a model. Run python scripts/verify_adapter.py --adapter /path/to/adapter.py --class-name MyAdapter.

  • If the adapter depends on a specific model API (e.g. GR00T's get_action(observation) with nested observations), pass a model stub file (defining a make_model_stub() factory function) via --stub-module: python scripts/verify_adapter.py --adapter gr00t_adapter.py --class-name Gr00tProtocolAdapter --stub-module stub_gr00t.py

4-Component Pattern

Every inference interface adaptation follows the same architecture. The model's inference logic stays untouched — all protocol-related logic lives in separate files you create.

text

┌───────────────────────────────────────────────────┐
│  1. Entry point (CLI)                              │
│     - parse args, assemble components              │
├───────────────────────────────────────────────────┤
│  2. Transport server                               │
│     - WSS + msgpack, handshake, request loop,      │
│       error handling                               │
├───────────────────────────────────────────────────┤
│  3. Protocol adapter                               │
│     - convert observations: flat format -> model   │
│       format                                       │
│     - convert actions: model format -> flat format │
├───────────────────────────────────────────────────┤
│  4. Model inference interface (existing, reused)   │
│     - predict / get_action -> actions              │
│     - metadata, normalization                      │
├───────────────────────────────────────────────────┤
│  5. Model + weights (existing, untouched)          │
└───────────────────────────────────────────────────┘

Layers 4-5 already exist in the codebase. You create layers 1-3.

Generality Layering (Reuse vs Customize)

The components differ greatly in generality — do not blindly copy templates:

Component	Generality	Handling
Transport server (server_template.py)	High — WSS/msgpack/handshake/error contract are model-agnostic	can copy and use directly
Test client (scripts/test_client.py)	High — auto-discovers contract from handshake metadata	can copy and use directly
Protocol adapter (adapter_template.py)	Low — the model's observation format/action output varies by model	reference the skeleton only; customize the _derive_* methods and conversion logic per model
Adapter verification (verify_adapter.py)	Low — depends on the model API	test adapter.infer() via "contract injection", without depending on the model API

Key judgment: if the model's inference interface differs greatly from the template's assumptions, rewrite the adapter + write your own verification, reusing only the transport server and test client.

Workflow

Step 1: Identify the Model's Inference Interface

Find the existing inference interface in the codebase. It usually exposes:

• A predict/infer method — takes an example dict, returns actions (typically batched (B, T, D), possibly already denormalized)
• Model metadata — action_dim, state_dim, image_keys, action_chunk_size
• Normalization access — how to get/set denormalization parameters

Record the model's internal format:

• Does the model take a flat example ({"image": ..., "state": ..., "lang": ...}) or nested observations ({"video": {...}, "state": {...}, "language": {...}})?
• What does the model return? ((B, T, D) ndarray? multiple action keys? already denormalized?)
• Which image keys does the model expect? Derive them from the model's config or framework, don't hardcode. Different models use different camera names (cam_high vs base_0_rgb vs camera_head).

Step 2: Create the Protocol Adapter

The adapter is the heart of the adaptation. It converts between the flat wire format and the model's internal format. Create a class that wraps the model's inference interface. See references/adapter_template.py for the full template — after copying it to the project, you only need to implement the five _derive_* methods (deriving image keys, state dim, state keys, action dim, and action chunk size from the model config/metadata). __init__ calls all of them, so every attribute the metadata property reads is initialized before the handshake.

The adapter exposes two core interfaces:

python

class ProtocolAdapter:
    def __init__(self, model, **kwargs): ...      # derive and cache the model I/O contract
    def infer(self, obs: dict) -> dict:            # convert -> infer -> convert
        model_input = self._obs_to_example(obs)    # flat example or nested observation
        result = self._model.predict(examples=[model_input])
        return self._result_to_response(result, obs)
    @property
    def metadata(self) -> dict: ...                # handshake contract, see below

The model call inside infer() varies by model shape: flat-example models use model.predict(examples=[...]); nested-observation models (e.g. GR00T's get_action(observation)) call model.get_action(model_input) directly. See "Two Shapes of Model-Internal Observations" below for both shapes.

Why hierarchical metadata: groups related fields together (action, state, images, task) instead of a flat list. This makes the contract self-documenting, and clients can parse only the parts they need.

Observation conversion _obs_to_example — the most critical part. The wire format uses flat dot-notation keys that directly encode field paths, avoiding nested ambiguity. The conversion does three things:

1. Images — extract from the observation.images.{key} prefix, ordered by the model's expected image_keys; a missing required image key raises KeyError immediately.
2. Language — task → lang, using a fallback instruction when empty.
3. State — observation.state → (1, D), validating the dimension against state_dim; omitted when the model has no state input.

Action conversion _result_to_response — usually just squeeze the batch dimension: the model returns (B, T, D), validate batch=1, squeeze to (T, D), and output {"action": actions}.

Key decisions:

• Flat dot-notation wire format — use flat keys like observation.images.{key}. It avoids ambiguity and makes the wire contract easier to evolve.
• Image keys derived, not hardcoded — derive from the model's DataConfig (e.g. video_keys minus the video. prefix), not from a non-existent framework.image_keys. If unavailable, allow a CLI override via --image_keys.
• State validation — prevents silent garbage output from dimension mismatches.
• No server_timing in the adapter — the transport server adds timing because it is per-connection state. The adapter should be testable without a server.
• Batch squeeze — the model returns (B, T, D) but the protocol expects (T, D). Always squeeze and validate batch is 1.
• Model-internal field names configurable — the action field name in the model's return differs by model (action, actions, pred_actions, etc.); specify it via the model_action_key parameter, default "action". The wire field name stays fixed as action, never affected.
• Multi-action-key concatenation — some models (e.g. GR00T) return multiple action keys ({"x": ..., "y": ..., "z": ...}). Concatenate them per the model's action key order with np.concatenate([...], axis=-1) into (T, D), then output as action.

Two Shapes of Model-Internal Observations

The template assumes the model takes a flat example dict ({"image": ..., "state": ..., "lang": ...}). But many VLA models take nested observations (GR00T's get_action needs {"video": {...}, "state": {...}, "language": {...}}). _obs_to_example must implement the matching conversion per the model shape:

Shape	Model-side input	_obs_to_example produces
Flat example	{"image": (1,H,W,3), "state": (1,D), "lang": str}	assemble directly
Nested observation	{"video": {"cam": (B,T,H,W,C)}, "state": {"k": (B,T,D)}, "language": {key: [[str]]}}	assemble a nested dict from wire flat keys

The wire side (observation.images.{key} / observation.state / task) stays identical for both shapes; only the adapter-internal conversion differs.

Contract Derivation Reference Table (derive from model config/statistics, don't hardcode)

⚠️ Verification note: GR00T behavior is measured/verified; the rest are example references — each framework's config/statistics structure changes with version, so follow the actual model's config structure, don't copy paths blindly.

Framework/Model	image_keys source	state/action dim source
example1	policy.modality_configs['video'].modality_keys	policy.processor.state_action_processor.statistics (sum of dims per key)
example2	video keys in config	stats.json
example3	video items in meta/modality.json	meta/stats.json / relative_stats.json
Other	search config for camera key lists	search stats files containing mean/std/min/max/q01

Derivation principle: use the sum of the actual statistics dims for state_dim/action_dim, not the padded dims of the model's input layer, otherwise normalization index mismatches occur.

Step 3: Create the Transport Server

The transport server implements the WSS communication loop. It contains no inference logic — it just moves bytes. See references/server_template.py for the full template.

At startup the server generates a self-signed certificate at runtime with the cryptography library (RSA 3072 + x509), builds an ssl.SSLContext(PROTOCOL_TLS_SERVER) (TLSv1.2+, restricted to secure cipher suites), then calls serve(..., ssl=ssl_context) to enable TLS. The certificate files are deleted immediately after being loaded into the SSLContext.

The fixed shape of the communication loop (5 steps):

1. Handshake — send metadata as the first frame after connect
2. Receive observation — recv + deserialize (no envelope, no routing)
3. Infer — call adapter.infer(obs)
4. Attach timing — write server_timing into the response
5. Send response — send + serialize

Key decisions:

• prev_total_time is a local variable — reset per connection. The first request always has prev_total_ms == 0; the second has the total time of the first request.
• No message routing — a simple recv->infer->send loop. No {type, request_id, payload} envelope; fundamentally simpler.
• Error contract fixed — always send a traceback.format_exc() string and close with INTERNAL_ERROR. Sending another format (e.g. a structured dict) makes the client fail to parse.
• compression=None, max_size=None — the common setting for large numpy array payloads. Compression can cause binary-data issues; max_size=None allows arbitrarily large messages.

Step 4: Create the Entry Point

A thin CLI script that assembles the model, adapter, and server:

python

def main(args):
    # 1. Load the model (existing inference interface)
    model = load_model(args.ckpt_path, device=args.device, **kwargs)

    # 2. Wrap with the protocol adapter
    adapter = ProtocolAdapter(
        model,
        send_state=not args.no_state,
        fallback_instruction=args.fallback_instruction,
        **protocol_specific_kwargs,
    )

    # 3. Log the contract for train/test consistency checks
    contract = adapter.metadata
    logging.warning(
        "Serving on port=%d. Image keys: %s, state_dim: %d, action_dim: %d",
        args.port,
        contract["images"]["keys"],
        contract["state"]["dim"],
        contract["action"]["dim"],
    )

    # 4. Start the transport server
    server = TransportServer(
        adapter, host=args.host, port=args.port, metadata=contract,
    )
    server.serve_forever()

Why log the contract at startup: if the server's image keys don't match training, inference will produce garbage output. This catches the most common deployment error: serving a checkpoint with a mismatched observation format.

Step 5: Create the Test Client

Use the built-in scripts/test_client.py to verify the server end-to-end. The client auto-discovers the server's I/O contract from the handshake metadata, so it works with any adapter without code changes. The client connects via a wss:// URI and, because the server uses a self-signed certificate, handshakes with ssl.CERT_NONE (skipping certificate verification).

The test client supports both hierarchical and flat metadata formats. It verifies:

• Actions shape is (T, D) (unbatched)
• First request server_timing.prev_total_ms == 0
• Subsequent requests server_timing.prev_total_ms > 0

For server-less unit tests of the adapter layer (observation conversion / image-key validation / state-dimension validation / metadata), use scripts/verify_adapter.py, as described under "Built-in Scripts" above.

Core Commands

Command	Purpose
python scripts/verify_adapter.py --adapter /path/to/adapter.py --class-name MyAdapter	server-less adapter unit tests (observation conversion, image keys, state dim, metadata)
python scripts/verify_adapter.py --adapter gr00t_adapter.py --class-name Gr00tProtocolAdapter --stub-module stub_gr00t.py	adapter tests with a model stub for model-API-dependent adapters
python scripts/test_client.py --host 127.0.0.1 --port 8000	smoke test with random images
python scripts/test_client.py --host 127.0.0.1 --port 8000 --image_dir /path/to/images --num_inferences 5 --verbose	end-to-end test with real images
python scripts/test_client.py --host 127.0.0.1 --port 8000 --benchmark	performance benchmark

Parameter Confirmation

Parameter	Description
--adapter / --class-name	Adapter Python file path and class name to test (verify_adapter.py)
--stub-module	Model stub file defining make_model_stub() (for model-API-dependent adapters)
--host / --port	Server hostname and port (test_client.py; default 127.0.0.1:8000)
--image_dir	Directory of images loaded by key name (falls back to random images)
--image_keys	Comma-separated camera key names (overrides server metadata)
--prompt	Language instruction (default "pick up the red cube")
--num_inferences / --benchmark	Number of inference calls; run a performance benchmark
model_action_key	Model-internal action field name (adapter constructor kwarg, default "action"); wire field stays action

File Structure

New files go under migrate_to_cloudrobo/deployment/ at the project root. Per the "Generality Layering": only build the adapter-layer files yourself; reuse the skill's templates/built-in scripts for the protocol layer (either copy or reference them):

text

migrate_to_cloudrobo/deployment/
├── {protocol}_adapter.py              # adapter (Step 2, customized per model)
├── server_{protocol}.py               # entry point (Step 4, assembly)
└── tools/
    ├── {protocol}_server.py            # transport layer (Step 3, copy server_template.py)
    └── (test_client.py directly references the skill's built-in script)

Naming convention: replace {protocol} with the protocol name (e.g. openpi_adapter.py, server_openpi.py, openpi_server.py). The test client directly uses the built-in scripts/test_client.py (auto-discovers the contract from metadata; no need to write one).

Common Pitfalls

1. Hardcoded image keys — different models use different camera names (cam_high vs base_0_rgb vs camera_head). Always derive from the model's DataConfig (e.g. video_keys minus the video. prefix). If the framework doesn't expose image keys, allow a CLI override via --image_keys.

2. Forgetting the handshake metadata — the client needs the contract (state_dim, image_keys, action_dim) to build valid requests. Without metadata, the client can only guess the format, causing silent failures.

3. Wrong error format — per the contract, send a traceback.format_exc() string frame on error. Sending another format (e.g. a structured error dict) causes the client to fail parsing.

4. Batch-dimension confusion — the model returns (B, T, D) but the protocol expects (T, D). Always squeeze the batch dimension in the adapter and validate batch is 1.

5. Normalization mismatch — the model interface should handle denormalization. The adapter only passes the correct normalization keys. Verify the adapter does not double-denormalize (applying statistics in both the model and the adapter).

6. Missing state validation — if the model was trained with a state input, the adapter should validate that the observation contains state with the correct dimension. A missing state key should raise a clear error rather than silently producing garbage actions.

7. Incompatible serializers — the wire serialization must be identical on both ends. The skill ships plain msgpack with a custom numpy extension (pack_array/unpack_array encoding __ndarray__ / __npgeneric__ dicts, same as the reference test_openpi.py). If the project previously used a different codec (e.g. msgpack-numpy), replace it on both the server and the client so the bytes stay compatible.

8. BFloat16 → NumPy conversion — if the model runs in bf16, output tensors cannot be converted to NumPy directly (TypeError: Got unsupported ScalarType BFloat16). Add .float() before .numpy():

   python

   actions = pred_actions.detach().cpu().float().numpy()  # bf16 → f32 → numpy
   

9. websockets_sync_client API mismatch — websockets.sync.client.connect() doesn't accept ping_interval or ping_timeout (those are async-API args). Only pass compression, max_size, open_timeout.

10. Relative imports in test scripts — if the test client uses relative imports (from . import <helper module>), it can't run as a script directly. Use absolute imports or sys.path.insert(0, parent_dir) for standalone execution.

Reference Documents

• references/adapter_template.py — protocol adapter template (Step 2, customize the _derive_* methods per model)
• references/server_template.py — transport server template (Step 3, WSS/msgpack/handshake/error contract)
• scripts/test_client.py — generic WSS test client (auto-discovers the contract from handshake metadata)
• scripts/verify_adapter.py — server-less adapter unit-test tool
• references/iam-policies.md — authentication & access control (no cloud credentials required; local server security notes)