Build a NiceGUI Chat with CometAPI and Page-Local History

Last reviewed: November 15, 2026 (Asia/Shanghai). References fetched November 14 UTC. Checked baseline: Python 3.12.3, NiceGUI 4.0.0, and HTTPX 0.28.1.

Direct answer

Build a NiceGUI chat with CometAPI by creating one conversation object inside each page builder, calling the documented Chat Completions endpoint from an async button handler, and committing the user-and-assistant pair only after accepting a complete text response. Keep the key in Python and render both sides with ui.chat_message(..., text_html=False).

The state location matters. NiceGUI’s versioned reference explains that module-level variables are shared across users, while local variables inside a page builder belong to that page load. A global messages = [] would therefore make the conversation shared. The example below creates its history, controls, and request-in-progress flag together for each page.

Who this is for

Use this pattern when adding a small conversational tool to an existing NiceGUI application or building a local Python browser interface. It assumes familiarity with virtual environments and process configuration. Earlier accepted exchanges reach the next model request; reloading the page starts a new conversation.

The baseline was checked with synthetic HTTP responses and NiceGUI’s Python user simulation. The simulation bypassed the unused CPU process pool because the writing sandbox denied semaphore access. No real browser test, authenticated CometAPI call, or deployment was performed; the live acceptance workflow below remains a check for your installation.

For an interface built around independent questions, compare the Panel workbench . This NiceGUI example deliberately sends accepted conversation history.

Key takeaways

Construct mutable history inside the page builder, with a fresh list for each conversation.
Build a candidate request without changing accepted history; commit the pair after success.
Disable Send and Clear while waiting, and enforce the busy check in Python too.
Keep networking asynchronous so other pages can continue using the event loop.
Bound accepted history, input, response bytes, and displayed reply length.
Use literal text rendering and fixed failure messages.

Install and configure the local application

Create a new project directory and install the checked versions:

python3 -m venv .venv
. .venv/bin/activate
python -m pip install "nicegui==4.0.0" "httpx==0.28.1"

Use Python 3.11 or newer; the checked interpreter was 3.12.3. Supply COMETAPI_KEY and COMETAPI_MODEL through your local secret configuration before launching. Select an exact current model identifier that supports ordinary text Chat Completions and is available to your account. The application reports missing setting names without printing their values.

The CometAPI chat reference documents POST /v1/chat/completions, bearer authentication, the messages array, and completion outcomes. This tutorial sends only model, messages, and stream: false. Review model-specific output controls before adding them. The example’s character and turn limits are application policy, not token limits or account spending controls.

Build the transport and accepted history

Save this as comet_chat.py:

import asyncio
import json
from dataclasses import dataclass, field
from typing import Awaitable, Callable

import httpx

ENDPOINT = "https://api.cometapi.com/v1/chat/completions"
MAX_PROMPT_CHARS = 2_000
MAX_REPLY_CHARS = 8_000
MAX_TURNS = 6
MAX_BODY_BYTES = 262_144


class ChatError(Exception):
    pass


def accepted_text(data):
    choices = data.get("choices") if isinstance(data, dict) else None
    if (not isinstance(data, dict)
            or data.get("object") != "chat.completion"
            or not isinstance(choices, list) or len(choices) != 1):
        raise ChatError("No accepted text reply.")
    choice = choices[0]
    message = choice.get("message") if isinstance(choice, dict) else None
    if (not isinstance(choice, dict)
            or choice.get("finish_reason") != "stop"
            or not isinstance(message, dict)
            or message.get("role") != "assistant"
            or message.get("refusal") is not None
            or message.get("tool_calls") not in (None, [])
            or message.get("function_call") is not None
            or not isinstance(message.get("content"), str)
            or not message["content"].strip()
            or len(message["content"]) > MAX_REPLY_CHARS):
        raise ChatError("No accepted text reply.")
    return message["content"]


async def fetch_reply(messages, key, model, transport=None):
    async with asyncio.timeout(30):
        async with httpx.AsyncClient(
            timeout=httpx.Timeout(connect=5, read=25, write=10, pool=5),
            follow_redirects=False, transport=transport,
        ) as client:
            async with client.stream(
                "POST", ENDPOINT,
                headers={"Authorization": f"Bearer {key}",
                         "Accept": "application/json"},
                json={"model": model, "messages": messages, "stream": False},
            ) as response:
                media = response.headers.get("content-type", "").split(";", 1)[0]
                if response.status_code != 200 or media.strip().lower() != "application/json":
                    raise ChatError("No accepted text reply.")
                body = bytearray()
                async for chunk in response.aiter_bytes():
                    if len(body) + len(chunk) > MAX_BODY_BYTES:
                        raise ChatError("No accepted text reply.")
                    body.extend(chunk)
    return accepted_text(json.loads(body))


AnswerFunction = Callable[[list[dict[str, str]]], Awaitable[str]]


@dataclass
class Conversation:
    messages: list[dict[str, str]] = field(default_factory=list)
    busy: bool = False

    async def submit(self, prompt: str, answer_fn: AnswerFunction):
        if self.busy:
            raise ChatError("A request is already running.")
        if not isinstance(prompt, str) or not prompt.strip():
            raise ChatError("Enter a question.")
        if len(prompt) > MAX_PROMPT_CHARS:
            raise ChatError("Use at most 2,000 characters.")
        if len(self.messages) >= MAX_TURNS * 2:
            raise ChatError("Six turns reached. Clear this conversation first.")
        candidate = [dict(message) for message in self.messages]
        candidate.append({"role": "user", "content": prompt})
        self.busy = True
        try:
            answer = await answer_fn(candidate)
            if (not isinstance(answer, str) or not answer.strip()
                    or len(answer) > MAX_REPLY_CHARS):
                raise ChatError("No accepted text reply.")
            self.messages = candidate + [{"role": "assistant", "content": answer}]
            return answer
        finally:
            self.busy = False

    def clear(self):
        if self.busy:
            raise ChatError("A request is already running.")
        self.messages.clear()

Conversation.submit copies accepted messages and adds the new user input to a candidate list. If the request fails, that candidate is discarded. On success, one assignment installs the candidate together with the assistant reply. The displayed transcript follows the same rule: an unsuccessful question stays in the input box for inspection, without becoming a conversation turn.

The parser accepts exactly one assistant text choice with finish_reason: "stop". It rejects truncation, refusals, current or legacy function-call output, empty content, and replies over 8,000 characters. These are local acceptance rules for a text conversation. A tool-using application needs a different result type and execution policy.

The HTTPX async guide documents the streaming context and byte iterator. Here, HTTP streaming controls how the ordinary JSON body is read; stream: false still selects a non-SSE model response. The code limits accumulated decoded bytes before JSON parsing. This does not establish a strict ceiling on HTTPX’s transient decompression allocations.

The timeout reference distinguishes connect, read, write, and pool waits. The outer asyncio timeout adds a cooperative 30-second budget around the network operation. Parsing follows the timeout block and is limited by the accepted body size. A timeout establishes that this caller stopped waiting, without proving the remote generation stopped.

Each low-volume submission owns and closes its client. For a busier application, give a reusable client an explicit application lifecycle, as discussed in the Python HTTPX client guide .

Wire the NiceGUI page

Save this beside the first file as app.py:

import os

from nicegui import ui

from comet_chat import ChatError, Conversation, fetch_reply


def required_setting(name):
    value = os.environ.get(name, "").strip()
    if not value:
        raise RuntimeError(f"Missing setting: {name}")
    return value


def build_page(answer_fn):
    conversation = Conversation()
    client = ui.context.client

    ui.label("Local conversation").classes("text-2xl")
    ui.label("Six accepted turns per page load. Clear starts again.")
    transcript = ui.column().classes("w-full max-w-3xl")
    entry = ui.input("Question").classes("w-full max-w-3xl")
    status = ui.label("Ready.")

    async def send():
        if conversation.busy:
            return
        prompt = entry.value or ""
        for control in controls:
            control.disable()
        status.set_text("Waiting for a complete reply...")
        try:
            answer = await conversation.submit(prompt, answer_fn)
        except ChatError as error:
            if not client.is_deleted:
                status.set_text(str(error))
        except Exception:
            if not client.is_deleted:
                status.set_text("Request failed; history is unchanged. No automatic retry.")
        else:
            if not client.is_deleted:
                with transcript:
                    ui.chat_message(prompt, name="You", sent=True, text_html=False)
                    ui.chat_message(answer, name="Assistant", text_html=False)
                entry.set_value("")
                status.set_text("Reply accepted.")
        finally:
            if not client.is_deleted:
                for control in controls:
                    control.enable()

    def clear():
        if conversation.busy:
            return
        conversation.clear()
        transcript.clear()
        entry.set_value("")
        status.set_text("Conversation cleared.")

    with ui.row():
        send_button = ui.button("Send", on_click=send)
        clear_button = ui.button("Clear", on_click=clear)
    controls = (entry, send_button, clear_button)


if __name__ == "__main__":
    KEY = required_setting("COMETAPI_KEY")
    MODEL = required_setting("COMETAPI_MODEL")

    @ui.page("/")
    def index():
        async def answer(messages):
            return await fetch_reply(messages, KEY, MODEL)

        build_page(answer)

    ui.run(host="127.0.0.1", port=8080, reload=False, show=False)

build_page runs once for every invocation of the decorated page function. Its Conversation() call gets a new list through default_factory=list. Keeping this construction inside the builder also makes the busy flag local to the same page. Opening a second tab creates a separate conversation, while a page reload resets its accepted turns.

The Send handler disables the controls before awaiting the provider. The conversation also checks busy before its first await, which rejects a second submission that reaches Python while the first is running. Clear observes the same state and cannot erase history during an in-flight request. These controls apply per page; several pages can still issue requests concurrently.

The NiceGUI chat-message implementation escapes text when text_html=False. A model reply containing HTML remains literal text in this configuration. The with transcript: block gives new messages an explicit parent, including after the asynchronous wait.

The client.is_deleted checks avoid updating a page that NiceGUI has already removed. They do not create an upstream cancellation guarantee. If the user leaves during a request, the remote result can remain uncertain. This implementation makes no automatic retry and exposes no Stop button.

Run the app:

python app.py

Open the local interface by entering 127.0.0.1:8080 in your browser. The versioned launch implementation documents host, port, reload, and show. Set the loopback address explicitly because the ordinary server default is 0.0.0.0. Keep any wider deployment behind an intentional access and spending policy.

Contract details to verify

Boundary	This application’s rule	Verification
State	One conversation per page load	Open two pages and inspect their mock request bodies.
History	Accepted pairs only	Fail a submission, then confirm its text is absent from the next request.
Concurrency	One pending request per page	Delay the mock answer and attempt Send and Clear.
Input	Nonempty text, at most 2,000 characters	Confirm rejected input never invokes the answer function.
Context	At most six accepted pairs	A seventh turn requires Clear.
Reply	One complete text result, at most 8,000 characters	Reject `length`, refusal, tool calls, and oversized text.
Transport	Fixed endpoint, no redirects, bounded JSON body	Check the exact URL, parsed payload, and rejection fixtures.
Rendering	Text with HTML disabled	Use a synthetic reply containing angle brackets and markup.

The six-turn policy bounds retained conversation size without silently discarding earlier context. At the maximum, the sixth request contains five accepted pairs and the new user turn. Character counts do not predict a model’s exact token usage; reduce these limits or add model-specific token admission when needed.

Happy-path operator workflow

First inject a harmless answer function into build_page so the interface can be checked without a key or generation. Have it return a fixed string containing markup. Verify one user-and-assistant pair appears after Send, the input clears on acceptance, and the markup appears as text.

For example, save this as fixture_app.py beside the two modules and run it with python fixture_app.py:

from nicegui import ui
from app import build_page


async def fixture_answer(messages):
    return "<strong>Fixture reply</strong>"


@ui.page("/")
def index():
    build_page(fixture_answer)


if __name__ == "__main__":
    ui.run(host="127.0.0.1", port=8080, reload=False, show=False)

Next supply an HTTPX mock through fetch_reply’s transport argument. Inspect the exact endpoint, bearer-header shape using a synthetic key, and parsed JSON body. After a second accepted turn, the request should contain three messages: the first user question, its assistant reply, and the new question. Open another page and verify its first request contains only its own question.

During preparation, offline checks passed for request construction, follow-up history, separate conversation objects, failed-turn rollback, overlap rejection, cancellation cleanup, input and turn limits, completion acceptance, and HTTP/JSON rejection. A NiceGUI simulated-user check also passed for callback display, fixed errors, failed-turn omission, Clear, and a separate page’s first request. These results establish local Python behavior within the simulation limits described above.

Finally configure your real key and model, start the app, and submit one short question. Confirm a complete reply, then ask a follow-up referring to it. Use Clear and confirm the next question starts fresh. Record versions, model identifier, time, and verdict without retaining credentials or conversation contents in routine diagnostics.

Error-path operator workflow

Make the injected answer function wait on a test event. Attempt a second submission and Clear while it waits. Neither should start another request or erase the pending conversation. Release the event and confirm the controls become available again.

Then return an HTTP error, redirect, HTML, malformed JSON, and an oversized body through the mock transport. None should become a chat pair. Also test valid JSON with finish_reason: "length", a refusal, and tool-call output. Follow each failure with a successful question and inspect its candidate history: the failed question must be absent.

Inject an unexpected exception containing a synthetic private marker. The page should show the fixed request-failed message without that marker. The application displays ChatError messages because all such messages in this implementation are fixed local strings; preserve that invariant when extending the class. Do not raise ChatError with raw provider bodies or exception text.

For the live error check, stop the development app, configure a deliberately unavailable model, restart, and submit a harmless question. Confirm the failure message and unchanged transcript. Restore the verified model before repeating the happy path. A manual resubmission is another provider attempt, so decide whether it is appropriate after an uncertain timeout.

Failure modes

Earlier answers appear on another page. Look for a conversation object or mutable history list moved to module scope. Keep construction inside the page builder; sharing the key and model configuration does not require sharing conversation state.

A failed question affects the next answer. Inspect where history is assigned. Only the accepted branch should install the candidate user message together with its assistant result. Status labels and rejected inputs must stay outside messages.

Other pages freeze while one request runs. Check for synchronous networking or CPU work added to the callback. NiceGUI’s versioned reference describes a shared event loop. An async function still blocks that loop when it calls a blocking library directly.

A complete-looking reply is rejected. Check completion outcome and local limits with synthetic data. The parser deliberately rejects length-limited and tool-call responses, even if they contain some text. Revisit the acceptance policy before changing how history is committed.

The seventh submission fails immediately. Six accepted turns have filled the local conversation. Clear starts again. This avoids silently changing the context while the user still sees the earlier transcript.

FAQ

Is page-local history authentication?

It scopes Python state to a page instance. It does not identify or authorize a person. The example binds to loopback; an externally reachable service needs its own access policy.

Does Clear delete provider records?

Clear resets this page’s accepted history and displayed transcript. The application sends no provider deletion request and creates no durable conversation database. Review retention separately for any logging or persistence you add.

Can this stream tokens?

NiceGUI can update elements as work progresses, but this example waits for a complete JSON response. Incremental output needs an SSE parser and an explicit policy for incomplete replies before those replies enter accepted history.

Why keep the failed question in the input?

It lets the user inspect or edit a question that never became an accepted turn. The application does not resubmit it automatically. Sending again starts a new request.

Sources checked

Seven distinct primary public references returned HTTP 200 on November 14, 2026 UTC. They support this evergreen implementation task. NiceGUI sources are pinned to the checked 4.0.0 release; the maintained provider, HTTPX, and Python references were checked against the methods used here.

CometAPI: Create a chat completion — route, authentication, messages, completion fields, and finish reasons.
NiceGUI 4.0.0: LLM Reference — page-local scope, shared globals, shared event loop, element contexts, and controls.
NiceGUI 4.0.0: ChatMessage implementation — text escaping, HTML option, author label, and sent-message presentation.
NiceGUI 4.0.0: ui.run implementation — listener address, port, browser opening, and reload settings.
HTTPX: Async Support — async client, streamed body iteration, response closure, and pooling tradeoffs.
HTTPX: Timeouts — connect, read, write, and pool waits.
Python: Coroutines and Tasks — cooperative timeout and cancellation behavior.

The commit policy, limits, fixed error messages, and operator procedures are application recommendations derived from those contracts.

Reader next step

Build the fixture-backed page, inspect a follow-up request, and verify that a failed question never enters the next request. Then Start with CometAPI for the account and model setup, keep the credential in Python, and run the same narrow acceptance workflow with your real configuration.