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

# vuer.server

## At

```python
class At
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/v0.0.52/vuer/server.py#L37)

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/v0.0.52/vuer/server.py#L42)

## VuerSession

```python
class VuerSession
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/v0.0.52/vuer/server.py#L52)

## VuerSession.__init__

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

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/v0.0.52/vuer/server.py#L53)

## VuerSession.socket

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

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/v0.0.52/vuer/server.py#L63)

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/v0.0.52/vuer/server.py#L101)

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

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

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/v0.0.52/vuer/server.py#L116)

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

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

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

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

Remove elements by keys.

Example Usage::

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

or a single key: ::

    app.remove @ "key1"

## VuerSession.popleft

```python
def popleft(self)
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/v0.0.52/vuer/server.py#L232)

## VuerSession.pop

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

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/v0.0.52/vuer/server.py#L238)

## VuerSession.clear

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

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/v0.0.52/vuer/server.py#L244)

clears all client messages

## VuerSession.stream

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

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/v0.0.52/vuer/server.py#L248)

## VuerSession.spawn_task

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

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/v0.0.52/vuer/server.py#L251)

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.

## Vuer

```python
class Vuer(PrefixProto, Server)
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/v0.0.52/vuer/server.py#L291)

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
free_port: bool = Flag('Kill what is running on the requested port if True.')
```

```python
static_root: str = Proto('.', help='root for file serving')
```

```python
queue_len: int = Proto(100, help='use a max length to avoid the memory from blowing up.')
```

```python
queries: Dict = Proto({}, help='query parameters to pass')
```

```python
cert: str = Proto(None, dtype=str, help='the path to the SSL certificate')
```

```python
key: str = Proto(None, dtype=str, help='the path to the SSL key')
```

```python
ca_cert: str = Proto(None, dtype=str, help='the trusted root CA certificates')
```

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

## Vuer.relay

```python
async def relay(self, request)
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/v0.0.52/vuer/server.py#L377)

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

```python
async def bound_fn(self, session_proxy: VuerSession)
```

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/v0.0.52/vuer/server.py#L414)

This is the default generator function in the socket connection handler

## Vuer.spawn

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

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/v0.0.52/vuer/server.py#L443)

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: None

## Vuer.bind

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

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/v0.0.52/vuer/server.py#L465)

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.
:return: None

## Vuer.get_url

```python
def get_url(self)
```

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

Get the URL for the Tassa.
:return: The URL for the Tassa.

## 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/v0.0.52/vuer/server.py#L504)

## 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/v0.0.52/vuer/server.py#L516)

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/v0.0.52/vuer/server.py#L553)

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/v0.0.52/vuer/server.py#L557)

## Vuer.uplink

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

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/v0.0.52/vuer/server.py#L566)

## Vuer.downlink

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

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/v0.0.52/vuer/server.py#L597)

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/v0.0.52/vuer/server.py#L672)

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/v0.0.52/vuer/server.py#L738)

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/v0.0.52/vuer/server.py#L758)

## Vuer.run

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

[Source](https://github.com/vuer-ai/vuer-docs/blob/docs/v0.0.52/vuer/server.py#L777)

### Inherited members

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

```python
domain = Proto('https://vuer.ai', help='default url of web client')
```

```python
host = Proto('localhost', help='set to 0.0.0.0 to enable remote connections')
```

```python
port = Proto(DEFAULT_PORT, help='port to use')
```

```python
cors = Proto('https://vuer.ai,https://staging.vuer.ai,https://dash.ml,http://localhost:8000,http://127.0.0.1:8000,*', help='domains that are allowed for cross origin requests.')
```

```python
verbose = Flag('Print the settings if True.')
```

## 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`