Vuer

vuer.server

At

python
class At

Source

Proxy Object for using the @ notation. Also supports being called direction, which supports more complex arguments.

At.init

python
def __init__(self, fn)

Source

BoundFn

python
class BoundFn

Source

Generic wrapper for decorators that bind functions and enable .start() method.

This class wraps a function and optionally starts the Vuer server. It provides a .start() method that can be called to start the server later.

Example::

@app.spawn() async def main(session): ...

Later, start the server

main.start()

BoundFn.init

python
def __init__(self, inst: 'Vuer', attr_name: str, start: bool=False)

Source

:param inst: Vuer instance :param attr_name: Name of the attribute to set on the Vuer instance :param start: Whether to start the server immediately

BoundFn.start

python
def start(self, **kwargs)

Source

Start the Vuer server with optional keyword arguments.

VuerSession

python
class VuerSession

Source

VuerSession.init

python
def __init__(self, vuer: 'Vuer', ws_id: int, queue_len=100)

Source

VuerSession.socket

python
def socket(self)

Source

Getter for the websocket object.

this is useful for closing the socket session from the client side.

Example Usage::

@app.spawn(start=True): async def main(session: VuerSession): print("doing something...") await sleep(1.0)

print("I am done! closing the socket.") session.socket.close()

VuerSession.grab_render

python
async def grab_render(self, ttl=2.0, **kwargs) -> ClientEvent

Source

Grab a render from the client.

:param quality: The quality of the render. 0.0 - 1.0 :param subsample: The subsample of the render. :param ttl: The time to live for the handler. If the handler is not called within the time it gets removed from the handler list.

VuerSession.get_webxr_mesh

python
async def get_webxr_mesh(self, key: str='webxr-mesh', ttl=2.0) -> ClientEvent

Source

Request WebXR mesh data from the client.

This method sends a GET_WEBXR_MESH RPC request to the client and waits for a response containing the detected environmental meshes from the WebXR AR session.

The response contains mesh data including vertices, indices, semantic labels, and transformation matrices for each detected mesh.

Usage Example::

from vuer import Vuer, VuerSession from vuer.schemas import WebXRMesh, Scene from asyncio import sleep

app = Vuer()

@app.spawn(start=True) async def main(session: VuerSession): session.set @ Scene( children=[WebXRMesh(key="webxr-mesh", stream=False)] )

await sleep(2) # Wait for meshes to be detected

Request mesh data on-demand

mesh_data = await session.get_webxr_mesh(key="webxr-mesh")

meshes = mesh_data.value.get('meshes', []) print(f"Retrieved {len(meshes)} meshes")

for mesh in meshes: vertices = mesh['vertices'] indices = mesh['indices'] semantic_label = mesh.get('semanticLabel', 'unknown') matrix = mesh['matrix']

print(f"Mesh: {len(vertices)/3:.0f} vertices, label={semantic_label}")

:param key: The key of the WebXRMesh component to query (default: "webxr-mesh") :param ttl: The time to live for the handler in seconds. If no response is received within this time, a TimeoutError is raised (default: 2.0) :return: ClientEvent containing mesh data in event.value['meshes'] :raises asyncio.TimeoutError: If the client doesn't respond within ttl seconds :raises AssertionError: If websocket session is missing

VuerSession.send

python
def send(self, event: ServerEvent) -> None

Source

Sending the event through the uplink queue.

VuerSession.rpc

python
async def rpc(self, event: ServerRPC, ttl=2.0) -> Union[ClientEvent, None]

Source

Send a ServerRPC event to the client and wait for a response through the session queue

:param event: The ServerRPC event to send. :param ttl: The time to live for the handler. If the handler is not called within the time it gets removed from the handler list. :return: ClientEvent

VuerSession.set

python
def set(self) -> At

Source

Used exclusively to set the scene.

the @SET operator is responsible for setting the root node of the scene.

Examples: proxy @ Set(Scene(children=[...]))

or

app.set @ Scene(children=[...])

VuerSession.update

python
def update(self) -> At

Source

Used to update existing elements. NOOP if an element does not exist.

Supports passing in a list of elements. (Thank God I implemented this... so handy! - Ge)

Example Usage::

app.update @ [element1, element2, ...]

VuerSession.add

python
def add(self) -> At

Source

Used to add elements to a specific parent.

Requires a parentKey, or treats the Scene root node as the default parent.

Example Usage::

app.add(element1, element2, ..., to=parentKey.)

or using the Scene root node as the default parent: ::

app.add @ element1

VuerSession.upsert

python
def upsert(self) -> At

Source

Upsert elements to a specific parent.

Requires a parentKey, or treats the Scene root node as the default parent.

Example Usage::

app.upsert(element1, element2, ..., to=parentKey.)

or using the Scene root node as the default parent: ::

app.upsert @ element1

VuerSession.remove

python
def remove(self) -> At

Source

Remove elements by keys.

Example Usage::

app.remove @ ["key1", "key2", ...]

or a single key: ::

app.remove @ "key1"

VuerSession.popleft

python
def popleft(self)

Source

VuerSession.pop

python
def pop(self)

Source

VuerSession.clear

python
def clear(self)

Source

clears all client messages

VuerSession.stream

python
def stream(self)

Source

VuerSession.spawn_task

python
def spawn_task(self, task, name=None)

Source

Spawn a task in the running asyncio event loop

Useful for background tasks. Returns an asyncio task that can be canceled.

.. code-block:: python :linenos:

async background_task(): print('\rthis ran once')

async long_running_bg_task(): while True: await asyncio.sleep(1.0) print("\rlong running background task is still running")

@app.spawn_task async def main_fn(sess: VuerSession):

Prepare background tasks here:

task = sess.spawn_task(background_task) long_running_task = sess.spawn_task(long_running_bg_task)

Now to cancel a running task, simply

.. code-block:: python :linenos:

task.cancel()

Todos

▫️ Add a way to automatically clean up when exiting the main_fn.

VuerSession.forever

python
async def forever(self)

Source

Keep the session alive indefinitely.

This is useful when you want to set up a scene and keep the server running without the session closing. The session will remain active until the client disconnects or the server is stopped.

Example Usage::

@app.spawn(start=True) async def main(session: VuerSession): session.set @ Scene(Box(args=[0.2, 0.2, 0.2], key="box")) await session.forever()

python
DEFAULT_CLIENT_ROOT: Path = Path(__file__).parent / 'client_build'

Vuer

python
class Vuer(Server)

Source

Vuer Server

This is the server for the Vuer client.

Usage::

app = Vuer()

@app.spawn async def main(session: VuerSession): session.set @ Scene(children=[...])

app.run()

.. automethod:: bind .. automethod:: spawn .. automethod:: relay .. automethod:: bound_fn .. automethod:: spawn_task .. automethod:: get_url .. automethod:: send .. automethod:: rpc .. automethod:: rpc_stream .. automethod:: close_ws .. automethod:: uplink .. automethod:: downlink .. automethod:: add_handler .. automethod:: _ttl_handler .. automethod:: run

python
domain: str = EnvVar @ 'VUER_DOMAIN' | 'https://vuer.ai'
python
client_url: Optional[str] = None
python
port: int = EnvVar @ 'VUER_PORT' | DEFAULT_PORT
python
web_port: int = None
python
workspace_path: str = ''
python
cors: str = EnvVar @ 'VUER_CORS' | DEFAULT_CORS
python
static_root: str = EnvVar @ 'VUER_STATIC_ROOT' | '.'
python
free_port: bool = False
python
queue_len: int = 100
python
queries: Dict = None
python
client_root: Path = DEFAULT_CLIENT_ROOT
python
verbose: bool = False

Vuer.ssl

python
def ssl(self) -> str

Source

Returns "s" if SSL is enabled, "" otherwise.

Use in URL construction: f"http{self.ssl}://" or f"ws{self.ssl}://"

Vuer.local_ip

python
def local_ip(self) -> str

Source

Get the local LAN IP address.

This is a well-known and safe approach for determining your local IP. It uses a UDP socket connection to determine the local IP address that would be used to reach external networks. No data is actually sent to the remote address.

:return: The local IP address as a string, or "127.0.0.1" if unavailable.

Vuer.relay

python
async def relay(self, request)

Source

This is the relay object for sending events to the server.

Todo: add API for specifying the websocket ID. Or just broadcast to all. Todo: add type hint

Interface: <uri>/relay?sid=<websocket_id>

:return:

  • Status 200
  • Status 400

Vuer.static_prefix

python
def static_prefix(self) -> 'Url'

Source

URL prefix for static files, accessible over the network.

Uses local_ip and respects SSL settings for network access (e.g., VR devices).

Vuer.localhost_prefix

python
def localhost_prefix(self) -> 'Url'

Source

URL prefix for static files, localhost only.

Use this for local development when network access is not needed.

Vuer.format_urls

python
def format_urls(self) -> list

Source

Generate all relevant URLs for display based on connection context.

Returns a list of tuples (label, url) for different connection modes. Intelligently handles:

  • Local development (localhost)
  • LAN connections
  • Remote vuer.ai connections
  • Port display (hides default ports)
  • WebSocket parameter inclusion (when needed)

:return: List of (label, url) tuples

Vuer.bound_fn

python
async def bound_fn(self, session_proxy: VuerSession)

Source

This is the default generator function in the socket connection handler

Vuer.spawn

python
def spawn(self, fn: SocketHandler=None, start=False)

Source

Bind the socket handler function fn to vuer, and start the event loop if start is True.

Note: this is really a misnomer.

:param fn: The function to spawn. :param start: Start server after binding :return: BoundFn instance that can be called later with .start()

Vuer.bind

python
def bind(self, fn=None, start=False)

Source

Bind an asynchronous generator function for use in socket connection handler. The function should be a generator that yields Page objects.

:param fn: The function to bind. :param start: Start server after binding :return: BoundFn instance that can be called later with .start()

Vuer.get_url

python
def get_url(self, host: str='localhost')

Source

Get the URL for the Vuer client.

:param host: The host to use in the websocket URL (e.g., "localhost" or IP address). :return: The URL for the Vuer client.

Vuer.send

python
async def send(self, ws_id, event: ServerEvent=None, event_bytes=None)

Source

Vuer.rpc

python
async def rpc(self, ws_id, event: ServerRPC, ttl=2.0) -> Union[ClientEvent, None]

Source

RPC only takes a single response. For multi-response streaming, we need to build a new one

Question is whether we want to make this RPC an awaitable funciton.

:param ttl: The time to live for the handler. If the handler is not called within the time it gets removed from the handler list.

Vuer.rpc_stream

python
async def rpc_stream(self, ws_id, event: ServerEvent=None, event_bytes=None)

Source

This RPC offers multiple responses.

Vuer.close_ws

python
async def close_ws(self, ws_id)

Source

Vuer.uplink

python
async def uplink(self, proxy: VuerSession)

Source

Vuer.downlink

python
async def downlink(self, request: Request, ws: WebSocketResponse)

Source

The websocket handler for receiving messages from the client.

:param ws: The websocket. :param request: The request (unused). :return: None

Vuer.add_handler

python
def add_handler(self, event_type: str, fn: EventHandler=None, once: bool=False) -> Callable[[], None]

Source

Adding event handlers to the vuer server.

:param event_type: The event type to handle. :param fn: The function to handle the event. :param once: Whether to remove the handler after the first call. This is useful for RPC, which cleans up after itself. The issue is for RPC, the key also needs to match. So we hack it here to use a call specific event_type to enforce the cleanup.

Usage:

As a decorator::

app = Vuer() @app.add_handler("CAMERA_MOVE") def on_camera(event: ClientEvent, session: VuerSession): print("camera event", event.etype, event.value)

As a function::

app = Vuer() def on_camera(event: ClientEvent, session: VuerSession): print("camera event", event.etype, event.value)

app.add_handler("CAMERA_MOVE", on_camera) app.run()

Vuer.socket_index

python
async def socket_index(self, request: BaseRequest)

Source

This is the relay object for sending events to the server.

Todo: add API for specifying the websocket ID. Or just broadcast to all. Todo: add type hint

Interface: <uri>/relay?sid=<websocket_id>

:return:

  • Status 200
  • Status 400

Vuer.add_route

python
def add_route(self, path, fn: Callable, method='GET', content_type='text/html')

Source

Vuer.run

python
def run(self, free_port=None, *args, **kwargs)

Source

Run the server.

.. deprecated:: Use :meth:start instead. This method will be removed in a future version.

Vuer.start

python
def start(self, free_port=None, *args, **kwargs)

Source

Vuer.loop_forever

python
async def loop_forever(self)

Source

Inherited members

Public imports

These symbols are available from this module. Their definitions are documented in the linked modules.