GRID Docs | General Robotics

from grid_cortex_client import CortexHubClient

Asynchronous WebSocket client for CortexHub. Provides a persistent WebSocket connection for efficient model inference. Supports both request-response and pub/sub patterns. For synchronous usage, use CortexClient instead.

Constructor

CortexHubClient(api_key: Optional[str] = None, base_url: Optional[str] = None, timeout: float = 60.0)

Methods

`available_models()`

available_models() -> list[str]

Return registered model identifiers.

`close()`

close() -> None

Close WebSocket connection and cleanup pub/sub resources.

`collect()`

collect(count: Optional[int] = None) -> list[HubResult]

Collect all results into a list. Convenience method that collects subscribe() results.

If provided, collect exactly this many results. If None, collect until no pending requests remain.

Returns: List of HubResult objects.

`connect()`

connect() -> None

Establish WebSocket connection.

`help()`

help(model_id: str) -> str

Return documentation for a model.

`publish()`

publish(model_id: Union[str, ModelType], request_id: Optional[str] = None, **kwargs: Any) -> str

Publish a model inference request (non-blocking). Sends the request and returns immediately. Results arrive via subscribe().

required

Model to run (ModelType enum or string)

Optional ID to track this request. Auto-generated if not provided.

Model-specific parameters (e.g., image_input, prompt)

Returns: The request_id (useful when auto-generated) Raises: RuntimeError: If not connected

`run()`

run(model_id: Union[str, ModelType], **kwargs: Any) -> Any

Run model inference via CortexHub (blocking request-response).

required

Model to run (ModelType enum or string)

Model-specific parameters (e.g., image_input, prompt)

Returns: Model output (same format as CortexClient.run) Raises: RuntimeError: If not connected or if pub/sub mode is active CortexHubError: If server returns an error

`submit()`

submit(model_id: Union[str, ModelType], request_id: Optional[str] = None, timeout_s: float | None = None, **kwargs: Any) -> 'asyncio.Task[HubResult]'

Submit a request and return a Task for its eventual result. This is a convenience wrapper over the pub/sub API that lets you write:

async with CortexHubClient() as hub: … depth_t = hub.submit(ModelType.ZOEDEPTH, image_input=img) … mask_t = hub.submit(ModelType.GSAM2, image_input=img, prompt=“cat”) … depth_res, mask_res = await asyncio.gather(depth_t, mask_t)

subscribe(count: Optional[int] = None) -> AsyncIterator[HubResult]

Subscribe to inference results. Yields results as they arrive from published requests.

If provided, stop after receiving this many results. If None, yields indefinitely until no pending requests remain.

Yields: HubResult objects containing request_id, model, data, ok, and error.

await hub.publish(ModelType.ZOEDEPTH, request_id="d1", image_input=img)
await hub.publish(ModelType.GSAM2, request_id="m1", image_input=img, prompt="cat")
async for result in hub.subscribe(count=2):
    if result.ok:
        print(f"{result.request_id}: shape={result.data.shape}")
    else:
        print(f"{result.request_id}: error={result.error}")

​Constructor

​Methods

​available_models()

​close()

​collect()

​connect()

​help()

​publish()

​run()

​submit()

​subscribe()