Build a Textual Terminal Client for CometAPI

Last reviewed: November 15, 2026 (Asia/Shanghai). Checked baseline: Python 3.12.3, Textual 8.2.8, and HTTPX 0.28.1.

Direct answer

Build a Textual terminal client for CometAPI by handling Input.Submitted, starting the network operation in a Textual worker, and displaying only an accepted complete text response. Keep a small submission gate in the interface: one request may be pending, while the input remains available for drafting the next question. Pressing Enter during that wait leaves the draft intact and sends nothing further.

That separation matters in an interactive terminal. Calling an async function directly from a message handler can still hold up that handler’s message processing. The Textual workers guide demonstrates this problem with network requests and shows how a worker lets other interface messages continue. This tutorial applies that mechanism to one explicit text submission.

Who this is for

Use this small application when you want a keyboard-driven local prompt tool with a scrollable answer area. Each question is independent: the next request contains only its submitted text. The application holds one visible answer and creates no conversation database.

You need Python 3.11 or newer, a terminal supported by Textual, and a CometAPI key and current model identifier supplied through process configuration. The checked code uses Python 3.12.3. For a browser interface with a visible exchange log, see the Panel prompt workbench .

Key takeaways

Dispatch on Enter through Input.Submitted; ordinary editing makes no request.
Start a worker without awaiting its completion in the input handler.
Set the pending flag before scheduling work so a second submission cannot slip through.
Preserve text drafted while a reply is pending.
Accept one complete assistant text choice before updating the answer area.
Test the interface with a delayed synthetic answer before connecting a live account.

Install and configure the local application

Create an isolated environment and install the checked application dependencies:

python3 -m venv .venv
. .venv/bin/activate
python -m pip install "textual==8.2.8" "httpx==0.28.1"

Supply COMETAPI_KEY and COMETAPI_MODEL through your local secret configuration before launch. The key belongs in the Python process, outside the source file and prompt field. Choose an exact current model ID that supports ordinary text Chat Completions and is available to your account. No model name embedded in an old tutorial establishes current access.

The CometAPI chat reference documents the endpoint, bearer authentication, messages, and completion outcomes used below. The request sends only model, one user message, and stream: false. Review the selected model’s requirements before adding optional sampling, reasoning, or output parameters.

Build the request boundary

Save this first block as terminal_app.py:

import asyncio
import json
import os

import httpx
from textual import work
from textual.app import App, ComposeResult
from textual.containers import VerticalScroll
from textual.widgets import Footer, Input, Static


ENDPOINT = "https://api.cometapi.com/v1/chat/completions"
MAX_PROMPT_CHARS = 4_000
MAX_RESPONSE_BYTES = 262_144


class ReplyError(Exception):
    pass


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


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 ReplyError("response_shape")
    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 not isinstance(message.get("content"), str)
        or not message["content"].strip()
    ):
        raise ReplyError("response_outcome")
    return message["content"]


async def fetch_answer(prompt, key, model, transport=None):
    async with asyncio.timeout(30):
        async with httpx.AsyncClient(
            timeout=25.0, 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": [{"role": "user", "content": prompt}],
                    "stream": False,
                },
            ) as response:
                if response.status_code != 200:
                    raise ReplyError("http_status")
                media = response.headers.get("content-type", "")
                if media.split(";", 1)[0].strip().lower() != "application/json":
                    raise ReplyError("media_type")
                body = bytearray()
                async for chunk in response.aiter_bytes():
                    if len(body) + len(chunk) > MAX_RESPONSE_BYTES:
                        raise ReplyError("response_size")
                    body.extend(chunk)
    return accepted_text(json.loads(body))

The acceptance function requires a chat.completion object with one assistant choice, finish_reason: "stop", and nonempty text. It rejects refusals, tool calls, and length-limited output. These are application rules for a text prompt tool; the provider’s broader response format supports other outcomes.

HTTPX’s async reference documents the async streaming context, byte iterator, and automatic response closure. HTTP streaming here controls body consumption; the model request still asks for complete JSON. The byte check limits the accumulated decoded body before parsing. It does not establish a strict ceiling on transient decompression memory.

The 30-second async budget surrounds client setup, the request, body reading, and closure. JSON parsing follows it with a bounded accepted body. This local timeout cannot establish whether remote generation finished. The character and byte limits are application choices, separate from token allowances and account spending controls.

This low-volume example creates a client inside each operation and closes it on exit. For an application that needs a shared connection pool, use an explicit application lifetime as described in the Python HTTPX client tutorial .

Connect submission to a worker

Append this block to the same file:

class PromptApp(App):
    BINDINGS = [("ctrl+q", "quit", "Quit")]
    CSS = """
    #status { height: auto; padding: 1; }
    #answer-pane { height: 1fr; padding: 1; }
    #answer { height: auto; }
    #prompt { dock: bottom; }
    """

    def __init__(self, answer_fn):
        super().__init__()
        self.answer_fn = answer_fn
        self.busy = False

    def compose(self) -> ComposeResult:
        yield Static("Enter one independent question; press Enter to send.",
                     id="status", markup=False)
        with VerticalScroll(id="answer-pane"):
            yield Static("", id="answer", markup=False)
        yield Input(placeholder="Question", id="prompt",
                    max_length=MAX_PROMPT_CHARS)
        yield Footer()

    def on_mount(self) -> None:
        self.query_one("#prompt", Input).focus()

    def on_input_submitted(self, event: Input.Submitted) -> None:
        prompt = event.value.strip()
        status = self.query_one("#status", Static)
        if self.busy:
            status.update("Still waiting. Your next draft has not been sent.")
            return
        if not prompt or len(event.value) > MAX_PROMPT_CHARS:
            status.update("Enter 1 to 4,000 characters.")
            return
        self.busy = True
        event.input.value = ""
        self.query_one("#answer", Static).update("")
        status.update("Waiting for a complete reply. You may draft another question.")
        self.generate(prompt)

    @work(group="generation", description="Single text request")
    async def generate(self, prompt):
        try:
            answer = await self.answer_fn(prompt)
            if not isinstance(answer, str) or not answer.strip():
                raise ReplyError("empty_answer")
        except asyncio.CancelledError:
            raise
        except Exception:
            self.query_one("#status", Static).update(
                "No accepted reply. Check configuration before sending again."
            )
        else:
            self.query_one("#answer", Static).update(answer)
            self.query_one("#status", Static).update("Reply ready.")
        finally:
            self.busy = False


def main():
    key = required_setting("COMETAPI_KEY")
    model = required_setting("COMETAPI_MODEL")

    async def answer_fn(prompt):
        return await fetch_answer(prompt, key, model)

    PromptApp(answer_fn).run()


if __name__ == "__main__":
    main()

The Input reference distinguishes Changed from Submitted and documents the character limit. This application handles only submission. Typing, clearing the field after an accepted send, and drafting during a wait cannot trigger generation by themselves.

The pending flag is set synchronously before generate() schedules its worker. While it is true, another Enter updates the status and returns without clearing the input. Completion changes the answer and status, leaving any new draft untouched. The finally branch releases the gate after success or failure.

Textual’s worker decorator makes the decorated method schedule work when called, so this handler calls self.generate(prompt) without await. Keep network I/O asynchronous inside the worker. Moving a blocking network library into an async def function would still block the event loop.

The work decorator reference documents a subtle diagnostic detail: its default description can include a representation of the method’s arguments. The explicit fixed description avoids putting the submitted question into that worker description. Treat debug consoles and custom logging separately; this setting is not a complete privacy policy for every framework diagnostic.

The answer uses Static(markup=False). The Static reference documents that markup parsing otherwise defaults to enabled. A reply such as [bold]example[/bold] therefore remains literal text in this widget. The application adds no Markdown renderer or ANSI conversion.

Check the interface without a live account

Save the following as test_terminal_app.py. The injected answer function waits until the test releases it, making the pending state reproducible:

import asyncio

import pytest
from textual.widgets import Input, Static
from terminal_app import PromptApp


@pytest.mark.asyncio
async def test_submission_preserves_the_next_draft():
    started = asyncio.Event()
    release = asyncio.Event()
    calls = []

    async def answer_fn(prompt):
        calls.append(prompt)
        started.set()
        await release.wait()
        return "[bold]Literal fixture[/bold]"

    app = PromptApp(answer_fn)
    async with app.run_test() as pilot:
        box = app.query_one("#prompt", Input)
        box.value = "First question"
        await pilot.pause()
        assert calls == []
        await pilot.press("enter")
        await asyncio.wait_for(started.wait(), 1)
        box.value = "Next draft"
        await pilot.press("enter")
        assert calls == ["First question"]
        assert box.value == "Next draft"
        release.set()
        await app.workers.wait_for_complete()
        await pilot.pause()
        answer = app.query_one("#answer", Static).visual
        assert answer.plain == "[bold]Literal fixture[/bold]"
        assert not answer.spans
        assert box.value == "Next draft"
        assert not app.busy

Install the test tools and run the check:

python -m pip install pytest pytest-asyncio
python -m pytest -q test_terminal_app.py

The Textual testing guide documents run_test() as a headless application context, Pilot.press() for keyboard interactions, and pause() for pending messages. A headless test exercises the message flow without taking over the terminal screen.

During preparation, 15 offline cases passed across the interface and HTTP boundary. They covered editing without dispatch, pending submission rejection, draft preservation, a subsequent send, fixed error display and recovery, shutdown cancellation, exact request construction, independent message bodies, and ten transport or completion rejection cases. A separate render check confirmed literal markup with no style spans. The included test provides the focused submission check; the broader matrix below describes the additional coverage to retain in your project. No authenticated CometAPI request or interactive terminal acceptance run was performed.

Contract details to verify

Boundary	Required behavior	Local evidence
Editing	No network dispatch before Enter	Set or type a draft and inspect the answer-function call list.
Pending submission	One active local attempt; second Enter preserves the draft	Hold the first answer with an event, then submit again.
Request	Fixed POST endpoint, bearer header, one user message, complete JSON	Inspect a synthetic HTTPX transport request.
Completion	One accepted assistant text result	Reject empty choices, refusal, tool calls, and `length`.
Response reading	Status, JSON media type, and byte ceiling precede parsing	Return HTTP errors, redirects, HTML, malformed JSON, and excess bytes.
Failure recovery	Fixed status text and released submission gate	Raise a synthetic exception, then complete the next submission.
Exit	Local worker receives cancellation	Wait inside a synthetic answer function and quit the app.

Happy-path operator workflow

Run the headless check first. Then start the configured application with python terminal_app.py and submit one short, harmless question. Confirm that the input starts focused, the status changes to waiting, and a complete answer appears in the scrollable area.

While waiting, draft another question and press Enter. The draft should remain, with a status explaining that it has not been sent. Once the first result arrives, press Enter deliberately to send the draft. Its body should contain only that question; earlier answers are unavailable as context. Keep any retained verification record to dependency versions, a fixed outcome category, and the local acceptance result.

Failure modes

The input looks active but Enter does nothing. Confirm that on_mount() focuses the Input and that the handler is named on_input_submitted. Test focus through Pilot before troubleshooting credentials.

The terminal becomes unresponsive during a request. Check that the handler schedules the decorated worker without awaiting its completion. Then inspect the worker for synchronous I/O or other blocking work.

A second Enter starts another request. Check that the pending guard runs before clearing the input or scheduling work. The gate is local to this application instance. Multiple instances can still make concurrent upstream requests.

The status reports no accepted reply. An HTTP error, timeout, unexpected media type, malformed body, or unaccepted completion can reach that fixed message. Reproduce those branches with a synthetic transport. Avoid exposing raw exception strings, response bodies, authorization headers, prompts, or replies in routine diagnostics.

Quit leaves uncertainty about upstream work. Textual’s workers guide says running workers are canceled when the app exits. The code re-raises cancellation and unwinds its HTTP contexts. That establishes local cleanup behavior; it does not prove that the remote generation or its usage stopped. Reconcile uncertain attempts before deliberately repeating a request.

Error-path operator workflow

Inject an answer function that raises an exception containing a synthetic private marker. Confirm that the terminal shows only the fixed failure status, the marker is absent, the answer area remains empty, and another deliberate submission works.

Next use an HTTPX mock transport to return the rejection cases in the table. Assert one attempt for each submission and no redirect following. Finally, hold a synthetic answer open and press Ctrl+Q; verify that its cleanup runs. Keep these checks separate from account access and real network behavior.

FAQ

Why keep the input enabled while waiting?

It lets the user prepare the next independent question. The submission gate prevents that draft from becoming another local attempt before the pending operation finishes.

Does the application send conversation history?

Each request contains one user message. The application replaces the visible answer on the next accepted send and maintains no history array.

Can I switch to streaming output?

Streaming needs a separate SSE parser and a policy for incomplete answers. Keep this version’s complete-response acceptance intact until you implement and test that different contract.

Does exiting undo a request?

Exiting cancels the local worker. Remote completion and usage require their own evidence; the interface provides no rollback guarantee and makes no automatic retry.

Sources checked

These seven maintained primary references were fetched successfully on November 14, 2026 UTC, November 15 in Asia/Shanghai. They support an evergreen implementation checked against the dependency baseline above; this article makes no release or incident claim.

CometAPI: Create a chat completion — endpoint, bearer authentication, request messages, response shape, and finish reasons.
Textual: Workers — message-handler concurrency, decorated workers, cancellation, and application lifetime.
Textual: work decorator — worker group and diagnostic description parameters.
Textual: Input — submitted messages, changed messages, and maximum input length.
Textual: Static — content updates and optional markup parsing.
Textual: Testing — headless application tests and Pilot interactions.
HTTPX: Async Support — asynchronous requests, streaming byte reads, and context-managed closure.

The single-pending-request policy, independent questions, acceptance rules, limits, and test matrix are application recommendations built on those interfaces.

Reader next step

Run the delayed-answer test, verify the fixed error path, and check worker cleanup first. Then configure a current model and make one short terminal exchange. Preserve the explicit Enter action and draft behavior as you extend the interface, and repeat the same checks whenever you change its request lifecycle.