Integrating on-premises AI

MultiTaction Canvus is self-hosted. Your Canvus Server runs on your own network, so you can connect it to the AI model your organisation already runs and approves, on that same network. Canvas content then travels only between your Canvus Server, a small worker program you run, and your model. None of it needs to leave your network.

This guide shows the pattern end to end with the REST API: a dedicated account, a stream of canvas changes, a call to your own model, and results written back to the canvas as notes, connectors, anchors and files.

How the pieces fit

flowchart LR
    W[Canvus clients and web client] <--> S[Canvus Server]
    S <-- "REST API over HTTPS<br/>?subscribe stream" --> B[Your worker]
    B <-- "HTTP" --> M[Your model endpoint]
    subgraph Your network
      S
      B
      M
    end

Canvus Server holds the canvases and serves the REST API on the same HTTPS port as the web client.
Your worker is a small service you write. It follows canvas changes, decides when to act, calls your model and writes the result back. It only makes outbound connections, so it needs no open inbound port.
Your model endpoint is whatever your organisation runs: a self-hosted model server, an internal AI gateway, or a cloud service you have approved.

Canvus Server does not call the model itself. Your worker does, so the choice of model, where it runs, what content the worker sends to it, and what gets logged all stay under your control. The same pattern works with any model, local or cloud. If you point the worker at a cloud service, canvas content goes wherever that service runs, and that is a choice you make and can audit.

1. Give the worker its own account

Create a service account for the worker, grant it edit on only the canvases it should work on, and create an access token for it. Authentication and service accounts gives the exact calls.

A dedicated account gives you three things:

The worker can only see and change what you granted it. Use view on canvases it should read but never change.
You can switch it off at once by deleting its token.
Administrators can filter the audit log with ?author_id= to see the actions logged for that account.

2. Read the canvas

Everything on a canvas is available as JSON.

curl -H "Private-Token: $TOKEN" \
  https://canvus.example.com/api/v1/canvases/$CANVAS/notes

Useful sources of context:

Call	What it gives your model
`GET /canvases/{id}/notes`	`text`, `title`, `background_color`, `location`, `size`, `parent_id` of every note
`GET /canvases/{id}/widgets`	Every widget of every type, with `widget_type`, position, size and `parent_id`; the list also includes connectors and the canvas background, which have no position
`GET /canvases/{id}/anchors`	Named zones (`anchor_name`) and their position and size
`GET /canvases/{id}/connectors`	Which widgets are linked (`src.id`, `dst.id`)
`GET /canvases/{id}/tables` and `.../tables/{id}/cells`	Table structure and cell content
`GET /canvases/{id}/browsers`	The `url` of each web page on the canvas
`GET /canvases/{id}/images/{id}/download`, `.../pdfs/{id}/download`	The original files, for models that read images or documents

Position tells your model how people have arranged their ideas. Two notes close together, or inside the same anchor zone, usually belong together. A widget's location is relative to its parent, given by parent_id.

3. Follow changes

Rather than polling, open a stream on the notes of a canvas:

GET /api/v1/canvases/{id}/notes?subscribe

The first message is every note. Each later message is an array with the note that changed. Merge them by id, as described in Streaming and change tracking.

Decide how people ask the worker for help. Some conventions that work with plain notes:

a note whose title starts with AI:;
a note in a specific colour;
a note dropped inside an anchor zone with an agreed name.

Wait until a note has stopped changing for a few seconds before acting on it, so the worker does not answer a half-typed question.

4. Call your model

Your worker sends the context to your model and reads the answer. The example below uses the chat completions request format that many self-hosted model servers and gateways accept. If yours uses a different format, change only this function.

import requests

MODEL_URL = "http://ai.internal.example.com:8000/v1/chat/completions"
MODEL_NAME = "your-approved-model"


def ask_model(question, context):
    body = {
        "model": MODEL_NAME,
        "messages": [
            {"role": "system", "content": "Answer briefly, for a sticky note."},
            {"role": "user", "content": f"Canvas notes:\n{context}\n\nQuestion: {question}"},
        ],
    }
    r = requests.post(MODEL_URL, json=body, timeout=120)
    r.raise_for_status()
    return r.json()["choices"][0]["message"]["content"]

5. Write the result back

As a note

Place the answer next to the note that asked for it. Pass the source note's parent_id so location uses the same coordinates.

def post_note(src, text):
    loc, size = src["location"], src["size"]
    body = {
        "title": "AI answer",
        "text": text,
        "background_color": "#E3F2FD",
        "parent_id": src["parent_id"],
        "location": {"x": loc["x"] + size["width"] * src["scale"] + 40, "y": loc["y"]},
    }
    r = canvus.post(f"{API}/canvases/{CANVAS}/notes", json=body)
    r.raise_for_status()
    return r.json()

To refine the answer later, PATCH /canvases/{id}/notes/{note_id} with new text. To withdraw it, DELETE the note.

On Canvus Server 26.10.1 or later, add "markdown": true to show the text rendered as Markdown, which suits headings and lists from a model. Older servers create the note without it.

As a connector

Link the answer to the question so people can see where it came from:

canvus.post(f"{API}/canvases/{CANVAS}/connectors", json={
    "src": {"id": question_note["id"]},
    "dst": {"id": answer_note["id"]},
})

As anchors and clusters

To group related notes, create a named zone with POST /canvases/{id}/anchors (anchor_name, location, size), then move notes into it with PATCH on their location. Moving people's notes changes their work, so ask before you do it, for example by acting only on canvases created for that purpose.

As files

To hand a generated document or image to the people on the canvas, upload it to the canvas's uploads folder. Canvus detects the file type and creates an image, video, PDF or generic file widget for it.

curl -X POST -H "Private-Token: $TOKEN" \
  -F 'json={"title": "Workshop summary"}' \
  -F 'data=@summary.pdf' \
  https://canvus.example.com/api/v1/canvases/$CANVAS/uploads-folder

See Uploads folder. To place a file at an exact position on the canvas instead, use the typed endpoints, such as POST /canvases/{id}/pdfs or POST /canvases/{id}/images, with location in the json part.

6. Do not answer yourself

The API does not record who created or changed a widget, so the worker sees its own notes arrive on the stream like anyone else's. Keep a set of the id values the worker has created and skip them. Also skip notes whose title marks them as the worker's output, in case the worker restarts and loses its memory.

A complete worker

This worker answers any note whose title starts with AI:. It uses only the calls shown above.

import json
import time
import threading
import requests

API = "https://canvus.example.com/api/v1"
TOKEN = "SERVICE_ACCOUNT_TOKEN"
CANVAS = "CANVAS_ID"

canvus = requests.Session()
canvus.headers["Private-Token"] = TOKEN
canvus.verify = True  # or the path to your server's CA certificate

notes = {}        # id -> note
mine = set()      # ids of notes this worker created
answered = set()  # ids of questions already answered
pending = {}      # id -> time of last change


def context_text():
    return "\n".join(f"- {n['title']}: {n['text']}"
                     for n in notes.values() if n["id"] not in mine)


def handle(note):
    question = note["text"].strip() or note["title"][3:].strip()
    answer = ask_model(question, context_text())
    reply = post_note(note, answer)
    mine.add(reply["id"])
    answered.add(note["id"])
    canvus.post(f"{API}/canvases/{CANVAS}/connectors",
                json={"src": {"id": note["id"]}, "dst": {"id": reply["id"]}})


def worker_loop():
    while True:
        now = time.time()
        for nid, changed in list(pending.items()):
            if now - changed > 3 and nid in notes:  # quiet for 3 s
                del pending[nid]
                try:
                    handle(notes[nid])
                except requests.RequestException as e:
                    print("failed:", e)
        time.sleep(1)


def on_note(note, first):
    if note.get("state") == "deleted":
        notes.pop(note["id"], None)
        return
    notes[note["id"]] = note
    if first or note["id"] in mine or note["id"] in answered:
        return
    if note["title"].startswith("AI:"):
        pending[note["id"]] = time.time()


def follow():
    while True:
        try:
            with canvus.get(f"{API}/canvases/{CANVAS}/notes?subscribe",
                            stream=True, timeout=(10, 60)) as r:
                r.raise_for_status()
                first = True
                for line in r.iter_lines():
                    if not line:
                        continue
                    msg = json.loads(line)
                    if first:
                        notes.clear()
                    for item in (msg if isinstance(msg, list) else [msg]):
                        on_note(item, first)
                    first = False
        except (requests.RequestException, ValueError) as e:
            print("stream ended:", e)
        time.sleep(2)


threading.Thread(target=worker_loop, daemon=True).start()
follow()

ask_model and post_note are the functions from steps 4 and 5. The worker ignores questions that already existed when it started (first), so a restart does not answer everything again. Store answered and mine on disk if you need them to survive a restart.

Summarise a canvas on demand

For a one-off job, no stream is needed. This collects the text of every note, asks the model for a summary and posts it as a new note:

NOTES=$(curl -s -H "Private-Token: $TOKEN" \
  https://canvus.example.com/api/v1/canvases/$CANVAS/notes \
  | jq -r '.[] | "- \(.title): \(.text)"')

SUMMARY=$(jq -n --arg notes "$NOTES" \
  '{model: "your-approved-model", messages: [{role: "user", content: ("Summarise these workshop notes:\n" + $notes)}]}' \
  | curl -s -H "Content-Type: application/json" -d @- \
      http://ai.internal.example.com:8000/v1/chat/completions \
  | jq -r '.choices[0].message.content')

jq -n --arg text "$SUMMARY" '{title: "Summary", text: $text}' \
  | curl -s -X POST -H "Private-Token: $TOKEN" -H "Content-Type: application/json" -d @- \
      https://canvus.example.com/api/v1/canvases/$CANVAS/notes

Operating notes

Throttle your own writes. Canvus Server does not rate-limit canvas API calls, so a busy worker can flood a canvas with notes. Limit how many answers it posts per minute.
Handle errors. 403 means the account lacks permission on that canvas. 402 means the server license is not valid. 400 explains a malformed request in msg. A dropped stream means "reconnect and resynchronise".
Search and filter in the worker. The API has no server-side search, so the worker filters the notes it receives.
No batch writes. Each widget is one request. Track the id values you create so a retry does not create duplicates.
Keep the worker close to the server. Calls and streams use the same HTTPS port as the web client. Trust your server's certificate in the worker (verify= in Python) rather than turning verification off.