Last reviewed: 2026-09-15

Direct answer

You can connect Semantic Kernel to CometAPI by creating an AsyncOpenAI client with CometAPI’s API base, then passing that client to OpenAIChatCompletion through its async_client argument. Semantic Kernel manages the chat abstraction and history while the injected client sends OpenAI-compatible chat-completion requests to CometAPI.

Two documented contracts make this possible. The CometAPI chat-completion reference identifies the compatible API base, required model and messages fields, and Server-Sent Events streaming behavior. The Semantic Kernel connector implementation shows that the Python service accepts an existing AsyncOpenAI instance.

Package releases change, and a source link that follows main is not a dependency pin. Resolve the packages in a disposable environment, verify the required interface, freeze the exact result, and commit that lock before deployment. Start with this requirements.in file:

semantic-kernel
openai

Resolve and test the interface once:

python -m venv .venv-resolve
. .venv-resolve/bin/activate
python -m pip install --upgrade pip
python -m pip install --requirement requirements.in
python -c "import inspect; from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion, OpenAIChatPromptExecutionSettings; from semantic_kernel.contents.chat_history import ChatHistory; p = inspect.signature(OpenAIChatCompletion).parameters; assert 'async_client' in p; assert hasattr(OpenAIChatCompletion, 'get_streaming_chat_message_content'); print('compatibility probe passed')"
python -m pip check
python -m pip freeze --all > requirements.lock

Do not deploy the resolver environment. Review and commit requirements.lock, then prove that it recreates cleanly on the same supported Python and platform family:

python -m venv .venv-run
. .venv-run/bin/activate
python -m pip install --requirement requirements.lock
python -m pip check

The checked-in lock, not the moving source branch, is the reproducibility boundary. Regenerate it deliberately when either library changes and rerun the compatibility probe before promotion.

Provide three runtime environment variables through your shell, deployment platform, or secret manager. COMETAPI_KEY holds the credential, COMETAPI_MODEL holds a model ID verified for the deployment, and COMETAPI_BASE_URL holds the documented CometAPI API base . Do not print the credential while checking configuration. If a support record must mention its value, replace the value with [REDACTED].

The example below injects the client, treats each streaming iteration as a possible collection of message contents, extracts only visible text, and commits the new history only after a successful turn. Assistant output goes to standard output; structured operational events go to standard error so the two streams do not corrupt each other.

import asyncio
from copy import deepcopy
import json
import os
import sys
import time

from openai import AsyncOpenAI
from semantic_kernel.connectors.ai.open_ai import (
    OpenAIChatCompletion,
    OpenAIChatPromptExecutionSettings,
)
from semantic_kernel.contents.chat_history import ChatHistory

MODEL_ID = os.environ['COMETAPI_MODEL']
SERVICE_ID = 'cometapi-chat'

SAFE_LOG_FIELDS = {
    'attempt',
    'elapsed_ms',
    'error_type',
    'http_status',
    'message_count',
    'model',
    'outcome',
    'output_characters',
    'request_id',
    'service_id',
}


def emit_event(event, **fields):
    record = {'event': event, 'provider': 'cometapi'}
    for name, value in fields.items():
        if name in SAFE_LOG_FIELDS and value is not None:
            record[name] = value
    print(json.dumps(record, sort_keys=True), file=sys.stderr, flush=True)


def message_contents_from(batch):
    if isinstance(batch, (list, tuple)):
        return batch
    return (batch,)


def visible_text_from(message_content):
    text_parts = []
    for item in getattr(message_content, 'items', ()) or ():
        text = getattr(item, 'text', None)
        if isinstance(text, str) and text:
            text_parts.append(text)

    if text_parts:
        return ''.join(text_parts)

    content = getattr(message_content, 'content', None)
    return content if isinstance(content, str) else ''


async def stream_turn(service, history, prompt):
    working_history = deepcopy(history)
    working_history.add_user_message(prompt)

    settings = OpenAIChatPromptExecutionSettings()
    started = time.monotonic()
    pieces = []
    wrote_output = False

    emit_event(
        'chat_started',
        attempt=1,
        model=MODEL_ID,
        service_id=SERVICE_ID,
        message_count=len(working_history),
    )

    try:
        response = service.get_streaming_chat_message_content(
            chat_history=working_history,
            settings=settings,
        )
        async for batch in response:
            for message_content in message_contents_from(batch):
                text = visible_text_from(message_content)
                if text:
                    pieces.append(text)
                    sys.stdout.write(text)
                    sys.stdout.flush()
                    wrote_output = True
    except Exception as exc:
        if wrote_output:
            sys.stdout.write('\n')
            sys.stdout.flush()
        emit_event(
            'chat_failed',
            attempt=1,
            elapsed_ms=int((time.monotonic() - started) * 1000),
            error_type=type(exc).__name__,
            model=MODEL_ID,
            outcome='error',
            service_id=SERVICE_ID,
        )
        raise

    if wrote_output:
        sys.stdout.write('\n')
        sys.stdout.flush()

    answer = ''.join(pieces)
    if not answer.strip():
        emit_event(
            'chat_failed',
            attempt=1,
            elapsed_ms=int((time.monotonic() - started) * 1000),
            error_type='EmptyStreamResult',
            model=MODEL_ID,
            outcome='error',
            service_id=SERVICE_ID,
        )
        raise RuntimeError('The stream completed without assistant text.')

    working_history.add_assistant_message(answer)
    emit_event(
        'chat_succeeded',
        attempt=1,
        elapsed_ms=int((time.monotonic() - started) * 1000),
        message_count=len(working_history),
        model=MODEL_ID,
        outcome='success',
        output_characters=len(answer),
        service_id=SERVICE_ID,
    )
    return answer, working_history


async def main():
    client = AsyncOpenAI(
        api_key=os.environ['COMETAPI_KEY'],
        base_url=os.environ['COMETAPI_BASE_URL'],
    )
    service = OpenAIChatCompletion(
        ai_model_id=MODEL_ID,
        service_id=SERVICE_ID,
        async_client=client,
    )

    history = ChatHistory()
    history.add_system_message(
        'Answer concisely and state when information is uncertain.'
    )

    try:
        _, history = await stream_turn(
            service,
            history,
            'Explain bounded retries in two short sentences.',
        )
    finally:
        await client.close()


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

A concrete happy-path operator workflow is:

  1. Confirm that the three environment variables exist without displaying their values.
  2. Verify the configured model ID and API base against the current CometAPI documentation.
  3. Install from the committed lock in a clean environment and run python -m pip check.
  4. Run python app.py 1>answer.txt 2>events.jsonl with a short, non-sensitive prompt.
  5. Confirm that answer.txt contains only assistant text and events.jsonl contains separate chat_started and chat_succeeded JSON records.
  6. Confirm that message_count grows after the returned assistant response is appended.
  7. Call stream_turn again with the returned history and a follow-up question to verify conversation continuity.
  8. Promote the same lock and configuration pattern to a limited deployment before increasing traffic.

Who this is for

This pattern is for Python developers who already use Semantic Kernel’s chat abstractions but want CometAPI to provide the OpenAI-compatible transport. It is useful when an application needs Semantic Kernel conversation state, plugins, or orchestration without replacing the framework’s chat service interface.

It is not a promise that every model accepts every Semantic Kernel option. Model capabilities and request parameters still need verification. If you are choosing a model ID for a reproducible integration, use the model-catalog verification guide before adding it to runtime configuration.

Key takeaways

  • Inject a configured AsyncOpenAI instance through OpenAIChatCompletion(async_client=...).
  • Keep the API base, credential, and model ID in runtime configuration rather than source code.
  • Resolve dependencies once, verify the required imports and methods, then deploy only from the committed lock.
  • Treat a streaming iteration as a possible collection and extract text from each message content instead of stringifying its container.
  • Preserve the same completed ChatHistory for follow-up turns, but do not commit a partially completed turn.
  • Send assistant text and JSON operational events to separate output streams.
  • Log identifiers, counts, timing, normalized status, and error classes—not prompts, message bodies, headers, credentials, or raw exceptions.

Sources checked

Four public sources define the boundaries used here:

These sources support the integration seam and request lifecycle. They do not establish that every CometAPI model supports every optional parameter, tool pattern, or modality.

Contract details to verify

Client construction. The injected client must receive the CometAPI base and runtime credential. Semantic Kernel’s constructor still needs a nonempty model ID when an existing asynchronous client is supplied. Keep one clear owner for the client and close it when the application shuts down.

Dependency identity. Record the exact Python version, operating-system family, and complete package lock used by the build. The compatibility probe checks the imports, constructor argument, and streaming method on which the example depends. Run it before freezing and again after every intentional lock update.

Endpoint family. This tutorial uses chat completions, not the Responses endpoint or a provider-native request shape. Before rollout, compare the configured base and endpoint family with the base-URL verification checklist .

Message roles. Put application instructions in an instruction message and end-user text in a user message. Retain prior assistant messages when continuing a conversation. If tools are added later, each tool result must correspond to its originating call; do not fabricate or reuse call identifiers.

Streaming shape. CometAPI documents incremental SSE chunks, including a possible final usage chunk without a normal text choice. Semantic Kernel presents streamed message contents through its own abstraction. The example therefore unwraps a returned batch, inspects each message content’s text items, and uses a string-valued content field only as a fallback. It never calls str() on the batch or message object, which could save a representation instead of the assistant’s visible delta.

History ownership. ChatHistory is mutable conversation state. The example deep-copies the accepted history for each turn and returns the working copy only after success. If the stream fails, the caller retains the last completed history. Keep a separate history for each conversation and define a reduction policy because unbounded histories increase work and can exceed a model’s context capacity.

Model-specific settings. The CometAPI reference distinguishes parameters by model family. Begin with default execution settings, then add only parameters supported by the selected model. Revalidate the request whenever the model ID or either client library changes. The request-body review guide is a useful companion before adding optional fields.

Safe observability. The example allowlists event, provider, service_id, model, attempt, elapsed_ms, message_count, output_characters, outcome, http_status, request_id, and error_type. A production adapter may populate status and request identifiers when its exception interface exposes them safely. It should not serialize prompts, completions, full histories, headers, environment values, or raw exceptions.

Failure modes

Missing configuration. If the model ID or credential is absent, initialization should fail before a request is sent. Treat that as a deployment configuration error. Do not substitute a guessed model or print the environment to diagnose it.

Dependency drift. An import failure, missing async_client argument, or missing streaming method means the resolved package combination no longer matches the example. Stop the build, restore the last working lock, or update the integration deliberately. Do not let a deployment resolve fresh package versions at startup.

Bad request. The CometAPI reference shows client errors for missing model or message data. A 400-class result calls for inspecting the sanitized request shape, selected model, roles, and supported parameters. Retrying an unchanged malformed request only adds noise.

Authentication rejection. A 401-class result means the runtime authentication configuration was rejected. Confirm that the intended secret is mounted in the correct deployment, replace it if necessary, and keep its contents out of logs and tickets.

Rate limiting. The CometAPI guidance associates 429 responses with exponential backoff. Use a bounded retry count, add jitter, and honor cancellation or deadline limits. Record the attempt number and final outcome. Never turn an interactive request into an unbounded retry loop.

Server or transport interruption. A server error or broken connection can end a stream after partial text has reached the terminal or user interface. Mark that turn incomplete and do not commit its working history. Retry only when the operation and user experience make repetition safe.

Empty or metadata-only completion. A streamed batch may have no visible text. The example ignores metadata-only contents and rejects a completed turn whose accumulated answer is empty. Investigate the selected model, finish condition, filtering outcome, and request configuration before treating it as success.

Output-channel mixing. Writing JSON records to the same unterminated stream as assistant deltas produces invalid line-delimited logs and contaminates user output. Keep assistant text on standard output, events on standard error, and capture them separately. The example terminates partial or complete assistant output with a newline before emitting the final event.

History divergence. If visible text is never added to ChatHistory, the next turn lacks the answer the user saw. Conversely, committing an interrupted answer overstates what completed. Use the returned working history only after success; retain the previous history on failure.

A concrete error-path operator workflow is:

  1. Stop automatic retries after the configured bound and preserve the failure timestamp.
  2. Keep the previous completed history; do not commit the failed working copy.
  3. Record only the model, service ID, attempt, elapsed time, normalized status, request ID when available, and error class.
  4. Classify the event as configuration, dependency, bad request, authentication, rate limit, server, transport, empty-stream, or output-routing failure.
  5. Repair configuration, dependency, 400-class, and 401-class causes before retrying; back off for 429; use a small retry budget for transient server or transport failures.
  6. Tell the caller when partial output was incomplete so a retry is not mistaken for a second independent answer.
  7. Escalate with sanitized fields, Python version, and locked package versions—never with the credential, headers, full prompt, full response, or raw exception object.

FAQ

Why inject a custom client instead of changing Semantic Kernel internals?

The Python constructor accepts an AsyncOpenAI client. Injection keeps provider configuration at an exposed boundary while preserving Semantic Kernel’s chat service interface.

Why does the example iterate twice inside the stream?

A streaming iteration can contain a collection of message contents. The outer loop receives the batch, and the inner loop extracts text from each content item. This avoids turning a list or framework object into user-visible text.

Does streaming automatically preserve the assistant response?

Do not assume it does. Accumulate visible text and add the completed assistant response to the working ChatHistory after success. Microsoft notes that final messages still need to be added to history.

Why are logs written to standard error?

It preserves standard output for assistant text and gives operators a clean JSONL event stream. Capture the two descriptors separately in a terminal, process manager, or container runtime.

Can I reuse one history for every user?

No. Use a separate history for each conversation boundary. Sharing mutable history can mix context between users and create correctness and privacy problems.

Should every failure be retried?

No. Fix malformed requests, dependency drift, and authentication configuration first. Apply bounded backoff to rate limits and carefully selected transient failures. A streaming request may already have exposed partial text, so retries must be visible and safe.

Why does the article generate a lock instead of publishing guessed version numbers?

The required package release numbers are not part of the documented API contract. Resolving in a controlled environment, running the compatibility probe, and committing the complete frozen result creates an exact project-specific dependency identity without inventing unsupported pins.

Reader next step

Create a disposable environment, run the compatibility probe, commit the resulting dependency lock, and reproduce it in a clean environment. Then use a single non-sensitive prompt to confirm that assistant text and operational events land in separate files. Make one follow-up call with the returned history, followed by one deliberately invalid model setting to rehearse the sanitized error workflow.

When those checks pass, Start with CometAPI , provision runtime configuration through your secret manager, and repeat the smoke test in a limited deployment.