Vuer

vuer.server

workspace_handler

python
async def workspace_handler(request, workspace: Workspace)

Source

Handle workspace file requests.

First checks for dynamic links, then resolves files through the workspace.

:param request: The aiohttp request object. :param workspace: The Workspace instance to resolve files from. :return: aiohttp Response object.

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

SceneOps

python
class SceneOps

Source

Base class providing scene graph operations (set, update, add, upsert, remove).

Subclasses must implement matmul to handle event dispatch. Used by both VuerSession and SceneStore.

SceneOps.set

python
def set(self) -> At

Source

Set the scene. Usage: obj.set @ Scene(...)

SceneOps.update

python
def update(self) -> At

Source

Update existing elements. Usage: obj.update @ element or obj.update @ [elem1, elem2]

SceneOps.add

python
def add(self) -> At

Source

Add elements. Usage: obj.add @ elem or obj.add(to="parent") @ elem

SceneOps.upsert

python
def upsert(self) -> At

Source

Upsert elements. Usage: obj.upsert @ elem or obj.upsert(to="parent") @ elem

SceneOps.remove

python
def remove(self) -> At

Source

Remove elements by key. Usage: obj.remove @ "key" or obj.remove @ ["k1", "k2"]

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(SceneOps)

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.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.till

python
async def till(self, event: str, timeout: float=None) -> ClientEvent

Source

Wait for and return an event of the specified type.

This method registers a one-time handler for the specified event type and awaits its arrival. Useful for waiting on specific events like INIT.

Example Usage::

@app.spawn(start=True) async def main(session: VuerSession):

Wait for the INIT event from the client

e = await session.till("INIT") client_type = e.value.get('clientType') # 'python' or browser info

if client_type == 'python': print("Python client connected!") else: print(f"Browser connected: {e.value.get('userAgent')}")

:param event: The event type to wait for (e.g., "INIT", "CAMERA_MOVE") :param timeout: Optional timeout in seconds. Raises asyncio.TimeoutError if exceeded. :return: The ClientEvent of the specified type :raises asyncio.TimeoutError: If timeout is specified and exceeded

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

Inherited members

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
workspace: Union[str, Path, List[Union[str, Path]], Workspace] = EnvVar @ 'VUER_WORKSPACE' | '.'
python
static_root: Union[str, Path, List[Union[str, Path]]] = None
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.create_webrtc_stream

python
def create_webrtc_stream(self, stream_id, track=None, codec='H264', max_bitrate=None, max_framerate=None, resolution=None)

Source

Create a WebRTC stream hosted on this Vuer server.

Must be called before app.start(). Routes are registered during startup.

Args: stream_id: Unique identifier for the stream (used in URL path). track: Optional custom MediaStreamTrack. If None, creates an internal track and enables push_frame(). codec: Preferred codec ("H264" or "VP8"). Default "H264". max_bitrate: Maximum bitrate in bps (e.g. 2_000_000 for 2 Mbps). max_framerate: Maximum frames per second (e.g. 15, 30). resolution: Tuple of (width, height) for fallback frame. Default (640, 480).

Returns: WebRTCStream instance with .push_frame() and .endpoint properties.

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.workspace_prefix

python
def workspace_prefix(self) -> 'Url'

Source

URL prefix for workspace 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 workspace 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, **filters)

Source

Register a spawn handler with optional client filtering.

Handlers are matched against the client's INIT event. Only the first matching handler runs; a warning is shown if multiple handlers match.

Filter syntax supports fnmatch wildcards:

  • client="python" - exact match
  • platform="*" - wildcard match

Example::

@app.spawn(client="python") async def python_handler(session: VuerSession):

Only for Python clients

...

@app.spawn(client="browser") async def browser_handler(session: VuerSession):

Only for browser clients

...

@app.spawn # No filter = matches all clients async def default_handler(session: VuerSession): ...

:param fn: The function to spawn. :param start: Start server after binding :param filters: Filter criteria to match against INIT event value (e.g., client="python", platform="Darwin") :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

Deprecated: Use await session.forever() instead.

.. deprecated:: 0.1.2 This method will be removed in a future version. Use await session.forever() for cleaner session-scoped waiting.

Inherited members

Public imports

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