---
title: "vuer.server"
section: "Python API"
order: 30
description: "Python API reference for vuer.server"
---

# vuer.server

## workspace_handler

```python
async def workspace_handler(request, workspace: Workspace)
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/server.py#L58)

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](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/server.py#L128)

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

## At.__init__

```python
def __init__(self, fn)
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/server.py#L133)

## SceneOps

```python
class SceneOps
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/server.py#L143)

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](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/server.py#L151)

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

## SceneOps.update

```python
def update(self) -> At
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/server.py#L156)

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

## SceneOps.add

```python
def add(self) -> At
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/server.py#L171)

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

## SceneOps.upsert

```python
def upsert(self) -> At
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/server.py#L186)

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

## SceneOps.remove

```python
def remove(self) -> At
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/server.py#L201)

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

## BoundFn

```python
class BoundFn
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/server.py#L216)

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](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/server.py#L232)

: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](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/server.py#L249)

Start the Vuer server with optional keyword arguments.

## VuerSession

```python
class VuerSession(SceneOps)
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/server.py#L297)

## VuerSession.__init__

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

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/server.py#L298)

## VuerSession.socket

```python
def socket(self)
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/server.py#L308)

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](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/server.py#L342)

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](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/server.py#L358)

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](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/server.py#L415)

Sending the event through the uplink queue.

## VuerSession.rpc

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

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/server.py#L436)

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](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/server.py#L468)

## VuerSession.pop

```python
def pop(self)
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/server.py#L474)

## VuerSession.clear

```python
def clear(self)
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/server.py#L480)

clears all client messages

## VuerSession.stream

```python
def stream(self)
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/server.py#L484)

## VuerSession.spawn_task

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

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/server.py#L487)

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](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/server.py#L523)

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](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/server.py#L568)

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

- `set` — from [`vuer.server.SceneOps`](/python-api/server#sceneops)
- `update` — from [`vuer.server.SceneOps`](/python-api/server#sceneops)
- `add` — from [`vuer.server.SceneOps`](/python-api/server#sceneops)
- `upsert` — from [`vuer.server.SceneOps`](/python-api/server#sceneops)
- `remove` — from [`vuer.server.SceneOps`](/python-api/server#sceneops)

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

## Vuer

```python
class Vuer(Server)
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/server.py#L591)

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](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/server.py#L708)

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](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/server.py#L716)

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](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/server.py#L735)

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](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/server.py#L777)

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](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/server.py#L814)

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](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/server.py#L822)

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](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/server.py#L829)

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](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/server.py#L914)

This is the default generator function in the socket connection handler

## Vuer.spawn

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

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/server.py#L943)

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](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/server.py#L993)

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](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/server.py#L1009)

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](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/server.py#L1025)

## Vuer.rpc

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

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/server.py#L1044)

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](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/server.py#L1081)

This RPC offers multiple responses.

## Vuer.close_ws

```python
async def close_ws(self, ws_id)
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/server.py#L1085)

## Vuer.uplink

```python
async def uplink(self, proxy: VuerSession)
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/server.py#L1096)

## Vuer.downlink

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

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/server.py#L1127)

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](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/server.py#L1240)

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](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/server.py#L1306)

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](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/server.py#L1326)

## Vuer.run

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

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/server.py#L1345)

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](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/server.py#L1361)

## Vuer.loop_forever

```python
async def loop_forever(self)
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/latest/src/vuer/server.py#L1470)

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

- `host` — from [`vuer.base.Server`](/python-api/base#server)
- `cert` — from [`vuer.base.Server`](/python-api/base#server)
- `key` — from [`vuer.base.Server`](/python-api/base#server)
- `ca_cert` — from [`vuer.base.Server`](/python-api/base#server)
- `WEBSOCKET_MAX_SIZE` — from [`vuer.base.Server`](/python-api/base#server)
- `REQUEST_MAX_SIZE` — from [`vuer.base.Server`](/python-api/base#server)

## Public imports

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

- [`Server`](/python-api/base#server) — `vuer.base.Server`
- [`handle_file_request`](/python-api/base#handle_file_request) — `vuer.base.handle_file_request`
- [`websocket_handler`](/python-api/base#websocket_handler) — `vuer.base.websocket_handler`
- [`Add`](/python-api/events#add) — `vuer.events.Add`
- [`ClientEvent`](/python-api/events#clientevent) — `vuer.events.ClientEvent`
- [`Frame`](/python-api/events#frame) — `vuer.events.Frame`
- [`GrabRender`](/python-api/events#grabrender) — `vuer.events.GrabRender`
- [`NullEvent`](/python-api/events#nullevent) — `vuer.events.NullEvent`
- [`Remove`](/python-api/events#remove) — `vuer.events.Remove`
- [`ServerEvent`](/python-api/events#serverevent) — `vuer.events.ServerEvent`
- [`ServerRPC`](/python-api/events#serverrpc) — `vuer.events.ServerRPC`
- [`Set`](/python-api/events#set) — `vuer.events.Set`
- [`Update`](/python-api/events#update) — `vuer.events.Update`
- [`Upsert`](/python-api/events#upsert) — `vuer.events.Upsert`
- [`Page`](/python-api/schemas/html_components#page) — `vuer.schemas.html_components.Page`
- [`Url`](/python-api/types#url) — `vuer.types.Url`
- [`Blob`](/python-api/workspace/workspace#blob) — `vuer.workspace.workspace.Blob`
- [`Workspace`](/python-api/workspace/workspace#workspace) — `vuer.workspace.workspace.Workspace`
- [`guess_content_type`](/python-api/workspace/workspace#guess_content_type) — `vuer.workspace.workspace.guess_content_type`
- [`workspace_from_config`](/python-api/workspace/workspace#workspace_from_config) — `vuer.workspace.workspace.workspace_from_config`