Build a CometAPI Python Client with HTTPX and asyncio

Last reviewed: November 5, 2026 (Asia/Shanghai); official references fetched November 4 UTC.

Direct answer

Build a small asynchronous CometAPI client around one reusable httpx.AsyncClient. Send a POST request to the documented /v1/chat/completions route, keep the key and model in trusted process configuration, set explicit connect, read, write, and pool timeouts, and validate the response before returning text. Keep the response body bounded before calling json.loads, and classify transport, HTTP, JSON, and completion-shape failures separately.

This article uses the standard Chat Completions contract and a text-only, non-streaming result. It accepts one completed assistant choice with non-empty content and rejects refusals, tool calls, and length-limited output. Those are application choices layered on the documented wire format. A streaming client or tool-using agent needs a separate parser and approval policy.

Who this is for

Use this pattern in an async Python service, worker, or command-line tool that needs a visible HTTP boundary without hiding transport policy inside an SDK. The examples assume Python 3.11 or newer so the asyncio.timeout context manager is available. Keep COMETAPI_KEY and COMETAPI_MODEL in the process environment or a secret manager. Never accept either value directly from a browser request, commit the key, or include it in exception text.

The timeout values and one-megabyte response ceiling below are local safeguards. They are not CometAPI account limits, latency guarantees, or billing rules. Choose values that fit the caller’s deadline and expected response size, then record the choice in your service configuration.

Key takeaways

• Reuse one httpx.AsyncClient for connection pooling and close it with async with during application shutdown.
• Set all four HTTPX timeout classes: connect, read, write, and pool. HTTPX applies timeouts by default, but explicit values make the contract reviewable.
• Use an outer asyncio.timeout for the whole operation; a read timeout alone does not cap queueing, request serialization, and response processing together.
• Check status and media type before parsing JSON, and return fixed error categories instead of an upstream error body.
• Count response chunks and reject an oversized body before deserialization.
• Validate the completion object, choice count, role, finish reason, refusal, tool calls, and non-empty content as one application contract.
• Test the request with httpx.MockTransport; a mock proves local construction and parsing, not credentials, model access, or production latency.

For a Rust comparison, see Build a CometAPI Rust Client with reqwest . For a focused response limit, see Cap CometAPI Response Bytes Before JSON Parsing in Node.js .

Define the client contract

Create comet_client.py with a narrow result and fixed error classes:

from __future__ import annotations

import asyncio
import json
import os
from dataclasses import dataclass
from typing import Any

import httpx


class CometError(Exception):
    """Base class for errors safe to map at an application boundary."""


class CometConfigError(CometError):
    pass


class CometTransportError(CometError):
    pass


class CometHTTPError(CometError):
    def __init__(self, status_code: int) -> None:
        self.status_code = status_code
        super().__init__(f"cometapi_http_{status_code}")


class CometResponseError(CometError):
    pass


@dataclass(frozen=True)
class ChatSettings:
    api_key: str
    model: str
    endpoint: str = "https://api.cometapi.com/v1/chat/completions"
    max_response_bytes: int = 1_048_576
    operation_timeout: float = 30.0


class CometChatClient:
    def __init__(self, settings: ChatSettings,
                 http: httpx.AsyncClient | None = None) -> None:
        if not settings.api_key.strip() or not settings.model.strip():
            raise CometConfigError("missing_cometapi_configuration")
        if settings.max_response_bytes < 1 or settings.operation_timeout <= 0:
            raise CometConfigError("invalid_cometapi_limits")
        self.settings = settings
        self.http = http or httpx.AsyncClient(
            timeout=httpx.Timeout(
                connect=5.0, read=25.0, write=10.0, pool=5.0
            ),
            follow_redirects=False,
        )
        self._owns_http = http is None

    async def aclose(self) -> None:
        if self._owns_http:
            await self.http.aclose()

    async def complete(self, prompt: str) -> str:
        if not isinstance(prompt, str) or not prompt.strip() or len(prompt) > 8_000:
            raise CometConfigError("invalid_prompt")
        payload = {
            "model": self.settings.model,
            "messages": [{"role": "user", "content": prompt}],
            "stream": False,
        }
        try:
            async with asyncio.timeout(self.settings.operation_timeout):
                async with self.http.stream(
                    "POST",
                    self.settings.endpoint,
                    headers={
                        "Authorization": f"Bearer {self.settings.api_key}",
                        "Content-Type": "application/json",
                        "Accept": "application/json",
                    },
                    json=payload,
                ) as response:
                    if not 200 <= response.status_code < 300:
                        raise CometHTTPError(response.status_code)
                    media_type = response.headers.get("content-type", "").split(";", 1)[0].strip().lower()
                    if media_type != "application/json":
                        raise CometResponseError("unexpected_response_media_type")
                    chunks: list[bytes] = []
                    total = 0
                    async for chunk in response.aiter_bytes():
                        total += len(chunk)
                        if total > self.settings.max_response_bytes:
                            raise CometResponseError("response_too_large")
                        chunks.append(chunk)
                    body = b"".join(chunks)
        except (httpx.TimeoutException, httpx.TransportError) as exc:
            raise CometTransportError("cometapi_transport_failed") from exc
        except TimeoutError as exc:
            raise CometTransportError("cometapi_operation_timeout") from exc
        try:
            data: Any = json.loads(body)
        except (json.JSONDecodeError, UnicodeDecodeError) as exc:
            raise CometResponseError("response_json_invalid") from exc
        return _accepted_text(data)


def _accepted_text(data: Any) -> str:
    if not isinstance(data, dict) or data.get("object") != "chat.completion":
        raise CometResponseError("unexpected_chat_completion_shape")
    choices = data.get("choices")
    if not isinstance(choices, list) or len(choices) != 1:
        raise CometResponseError("unexpected_chat_completion_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 not isinstance(message.get("content"), str)
        or not message["content"].strip()
        or message.get("refusal") is not None
        or message.get("tool_calls") is not None
    ):
        raise CometResponseError("unaccepted_chat_completion")
    return message["content"]


def settings_from_environment() -> ChatSettings:
    return ChatSettings(
        api_key=os.environ["COMETAPI_KEY"],
        model=os.environ["COMETAPI_MODEL"],
    )

httpx.AsyncClient owns the network call, while ChatSettings owns configuration and local limits. Passing an existing client makes the transport injectable in tests and lets an application share one pool across requests. When the class creates the client itself, aclose releases it; an application that passes its own client retains shutdown ownership.

The endpoint and request fields follow the CometAPI Chat Completions reference. follow_redirects=False keeps an authenticated POST from silently moving to another host or path; the caller sees the redirect status as an HTTP error. The key is placed in a header at call time and never appears in the request body or a public setting file.

HTTPX documents four timeout classes. connect covers establishing the connection, read covers waiting for response chunks, write covers sending request chunks, and pool covers waiting for a connection slot. The outer asyncio.timeout adds a total operation budget. The except order matters: HTTPX timeout exceptions are transport exceptions, while the outer context raises the built-in TimeoutError after cancelling the operation.

The streaming context counts aiter_bytes() chunks and stops after the first byte over the configured ceiling, so an oversized response is rejected before json.loads. The context also closes the response on every path. For a compact client with a separately enforced small body limit, response.content can be used instead, but it should not replace the size policy for untrusted or large responses.

The parser deliberately accepts one ordinary text answer. A tool_calls finish reason, refusal, missing content, or length finish reason is an application outcome requiring a different result type. Do not turn any of those into a successful string merely because the HTTP status is 200.

Use it from an async application

Keep the client lifetime at the application boundary:

import asyncio

from comet_client import CometChatClient, settings_from_environment


async def main() -> None:
    client = CometChatClient(settings_from_environment())
    try:
        answer = await client.complete("Explain idempotency in one sentence.")
    except Exception as exc:
        # Map fixed CometError categories to your application response.
        raise RuntimeError("chat_request_failed") from exc
    else:
        print(answer)
    finally:
        await client.aclose()


if __name__ == "__main__":
    asyncio.run(main())

In a FastAPI or worker process, create the shared AsyncClient during startup and close it during shutdown instead of constructing one for every request. Pass a request-specific deadline into your application policy, but do not store a request object or request-scoped cancellation state in the long-lived client. A local cancellation or timeout says that this caller stopped waiting; it does not establish whether the upstream generation completed or whether usage was recorded.

Keep error mapping outside the transport class. A 401, 429, or 500 can be logged as a status category and correlation ID without copying the provider body into an API response. Configuration, media-type, JSON, and completion-contract errors are deterministic and should not be retried automatically. A timeout may leave the remote result uncertain, so any retry needs an attempt limit, total deadline, and replay policy chosen for the operation.

Test without a live request

HTTPX’s MockTransport accepts a handler and returns predetermined responses. Use it to assert request construction and error classification without a credential or network access:

import json
import httpx
import pytest

from comet_client import (
    ChatSettings, CometChatClient, CometHTTPError, CometResponseError,
)


@pytest.mark.anyio
async def test_request_and_response_contract() -> None:
    seen = {}

    def handler(request: httpx.Request) -> httpx.Response:
        seen["method"] = request.method
        seen["url"] = str(request.url)
        seen["auth"] = request.headers["authorization"]
        seen["body"] = json.loads(request.content)
        return httpx.Response(
            200,
            headers={"content-type": "application/json"},
            json={
                "object": "chat.completion",
                "choices": [{
                    "finish_reason": "stop",
                    "message": {
                        "role": "assistant", "content": "Fixture answer.",
                        "refusal": None,
                    },
                }],
            },
        )

    transport = httpx.MockTransport(handler)
    async with httpx.AsyncClient(transport=transport) as http:
        client = CometChatClient(
            ChatSettings("fixture-key", "fixture-model"), http=http
        )
        assert await client.complete("Explain idempotency.") == "Fixture answer."

    assert seen["method"] == "POST"
    assert seen["url"].endswith("/v1/chat/completions")
    assert seen["auth"] == "Bearer fixture-key"
    assert seen["body"] == {
        "model": "fixture-model",
        "messages": [{"role": "user", "content": "Explain idempotency."}],
        "stream": False,
    }


@pytest.mark.anyio
@pytest.mark.parametrize("status", [401, 429, 500])
async def test_http_status_is_classified(status: int) -> None:
    def handler(_: httpx.Request) -> httpx.Response:
        return httpx.Response(status, headers={"content-type": "application/json"}, json={})

    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http:
        client = CometChatClient(ChatSettings("fixture-key", "fixture-model"), http=http)
        with pytest.raises(CometHTTPError) as error:
            await client.complete("Hi")
    assert error.value.status_code == status


@pytest.mark.anyio
async def test_truncated_completion_is_rejected() -> None:
    def handler(_: httpx.Request) -> httpx.Response:
        return httpx.Response(
            200,
            headers={"content-type": "application/json"},
            json={"object": "chat.completion", "choices": [{
                "finish_reason": "length",
                "message": {"role": "assistant", "content": "partial", "refusal": None},
            }]},
        )

    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http:
        client = CometChatClient(ChatSettings("fixture-key", "fixture-model"), http=http)
        with pytest.raises(CometResponseError, match="unaccepted_chat_completion"):
            await client.complete("Hi")

The first test checks the method, exact route, authorization shape, parsed request body, and accepted response. It compares the parsed body rather than raw JSON text, so key ordering does not matter while unexpected fields still fail the equality assertion. The status test demonstrates that an HTTP error is classified before its body is parsed. Add fixtures for HTML, malformed JSON, empty choices, refusal, tool calls, and an over-limit response.

Run the suite with your project’s normal async test configuration, for example pytest -q. Keep fixture-key and fixture-model synthetic. A passing mock proves serialization, parser behavior, and error mapping; it does not prove account access, current model availability, network reachability, billing, or latency.

Contract details to verify

Before a live smoke test, read the CometAPI model catalog and copy an exact current model ID. The catalog is public and documents IDs, model type, features, endpoint hints, and optional context metadata. Catalog records can change, so a model name in a saved example is not permanent evidence of availability.

Failure modes

The request returns HTML or a redirect. Inspect status, content type, and the configured URL. Keep /v1 and the operation path aligned with the current reference, and leave redirect following disabled while diagnosing.

The pool times out under load. A pool timeout means the client could not obtain a connection slot within its budget. Review httpx.Limits, application concurrency, and the outer operation deadline. Raising the timeout without measuring queueing can hide a capacity problem.

A response is parsed after an oversized read. Check the body-limit branch before json.loads. The implementation uses a streaming loop that stops as soon as the accumulated bytes exceed the limit; HTTPX controls individual chunk sizes.

A tool call or refusal is shown as text. The documented completion shape gives those outcomes their own fields and finish reasons. Return a typed application result and require authorization before executing a tool.

A timeout is retried repeatedly. The remote result may be unknown. Set a maximum attempt count and total deadline, and decide whether replaying the prompt is safe before adding a retry.

The key appears in a traceback or log. Keep fixed exception messages, avoid logging request.headers or response.content, and use a correlation ID with status and error class instead. Review Keep CometAPI Keys Out of Tutorial Repositories before sharing examples.

FAQ

Does HTTPX’s default timeout protect the whole operation?

No. HTTPX applies connect, read, write, and pool timeouts to network phases. An outer asyncio.timeout gives the application a total budget that also covers scheduling and response handling.

Should I use response.json()?

Only after checking status, media type, and the response-size policy. response.json() is convenient, but it does not replace a byte ceiling or your application-level completion checks.

Can this client parse streaming responses?

No. Streaming uses Server-Sent Events and chat.completion.chunk objects, then a [DONE] marker. Add a separate incremental reader with event framing, cancellation, and size limits.

Is MockTransport an integration test?

It is a deterministic transport test. It proves your local request and parser behavior. Use one separately authorized live smoke test to check credentials, current model access, and the real network path.

Should every HTTP error be retried?

No. Authentication, validation, wrong paths, malformed output, and unsupported options are deterministic. For transient failures, define a bounded retry policy that accounts for uncertain remote outcomes and application side effects.

Reader next step

Run the mock transport tests first, then set COMETAPI_KEY and COMETAPI_MODEL through a local secret mechanism and make one harmless staging request. Record only the endpoint family, model identifier, HTTP status class, response-contract verdict, elapsed time, and redaction result. Keep the live observation separate from the fixture result. For a byte-limit implementation detail, compare the Node.js response-boundary guide , then recheck the current CometAPI reference before adding streaming, structured output, or tool calls.

Sources checked

• CometAPI: Create a chat completion — fetched November 4, 2026 UTC. Supports the /v1/chat/completions route, bearer authentication, request fields, non-streaming response shape, SSE streaming, finish reasons, refusals, and tool-call fields.
• CometAPI: List available models — fetched November 4, 2026 UTC. Supports public model IDs, model types, features, endpoint hints, and optional context metadata.
• HTTPX: Timeouts — fetched November 4, 2026 UTC. Supports default timeout behavior and the connect, read, write, and pool timeout classes.
• HTTPX: Transports — fetched November 4, 2026 UTC. Supports AsyncClient transport injection and MockTransport handlers for predetermined responses.
• Python: json library — fetched November 4, 2026 UTC. Supports json.loads, JSONDecodeError, and the recommendation to limit untrusted JSON size before parsing.
• Python: asyncio tasks — fetched November 4, 2026 UTC. Supports asyncio.timeout, cancellation, and conversion to TimeoutError outside the context manager.

These are current public technical references for an evergreen Python integration. No authenticated request, deployment, or production latency result is claimed here.