microsoft/semantic-kernel · error · NotImplementedError

The _inner_get_streaming_chat_message_contents method is not

Error message

The _inner_get_streaming_chat_message_contents method is not implemented.

What it means

Raised by the base ChatCompletionClientBase._inner_get_streaming_chat_message_contents stub. It is the default async-generator implementation meant to be overridden by each concrete connector; calling it means the subclass never provided a streaming chat handler.

Source

Thrown at python/semantic_kernel/connectors/ai/chat_completion_client_base.py:76

        raise NotImplementedError("The _inner_get_chat_message_contents method is not implemented.")

    async def _inner_get_streaming_chat_message_contents(
        self,
        chat_history: "ChatHistory",
        settings: "PromptExecutionSettings",
        function_invoke_attempt: int = 0,
    ) -> AsyncGenerator[list["StreamingChatMessageContent"], Any]:
        """Send a streaming chat request to the AI service.

        Args:
            chat_history: The chat history to send.
            settings: The settings for the request.
            function_invoke_attempt: The current attempt count for automatically invoking functions.

        Yields:
            streaming_chat_message_contents: The streaming chat message contents.
        """
        raise NotImplementedError("The _inner_get_streaming_chat_message_contents method is not implemented.")
        # Below is needed for mypy: https://mypy.readthedocs.io/en/stable/more_types.html#asynchronous-iterators
        if False:
            yield

    # endregion

    # region Public methods

    async def get_chat_message_contents(
        self,
        chat_history: "ChatHistory",
        settings: "PromptExecutionSettings",
        **kwargs: Any,
    ) -> list["ChatMessageContent"]:
        """Create chat message contents, in the number specified by the settings.

        Args:
            chat_history (ChatHistory): A list of chats in a chat_history object, that can be

View on GitHub (pinned to c028a0c7dc)

Solutions

  1. Implement 'async def _inner_get_streaming_chat_message_contents(self, chat_history, settings, function_invoke_attempt=0)' as an async generator in your subclass.
  2. If the provider is non-streaming only, call get_chat_message_contents instead.
  3. Use a shipped connector class that already implements streaming.

Example fix

# before
class MyService(ChatCompletionClientBase): ...
async for m in my_service.get_streaming_chat_message_contents(history, settings): ...
# after
class MyService(ChatCompletionClientBase):
    async def _inner_get_streaming_chat_message_contents(self, chat_history, settings, function_invoke_attempt=0):
        yield [StreamingChatMessageContent(role=AuthorRole.ASSISTANT, choice_index=0, items=[StreamingTextContent(choice_index=0, text="hi")])]
Defensive patterns

Strategy: type-guard

Validate before calling

def service_implements_streaming(service) -> bool:
    from semantic_kernel.connectors.ai.chat_completion_client_base import ChatCompletionClientBase
    method = type(service).__dict__.get("_inner_get_streaming_chat_message_contents")
    return callable(method) and method is not ChatCompletionClientBase._inner_get_streaming_chat_message_contents

Type guard

from semantic_kernel.connectors.ai.chat_completion_client_base import ChatCompletionClientBase

def overrides_streaming_chat(cls: type) -> bool:
    m = cls.__dict__.get("_inner_get_streaming_chat_message_contents")
    return callable(m)

Try / catch

try:
    async for chunk in service.get_streaming_chat_message_contents(history, settings):
        process(chunk)
except NotImplementedError as e:
    if "_inner_get_streaming_chat_message_contents" in str(e):
        raise NotImplementedError("Service subclass must implement _inner_get_streaming_chat_message_contents") from e
    raise

Prevention

When it happens

Trigger: A chat completion service subclass that does not override _inner_get_streaming_chat_message_contents, then get_streaming_chat_message_contents/get_streaming_chat_message_content is iterated. The 'if False: yield' block exists only to satisfy mypy's async-iterator requirement.

Common situations: A custom connector that implemented only the non-streaming method; a service that only supports non-streaming but the caller used the streaming API; a mock subclass missing the streaming override.

Related errors


AI-assisted analysis of microsoft/semantic-kernel@c028a0c7dc (2026-08-13). Data as JSON: /api/errors/0f6448f9d3b3f8f8. Report an issue: GitHub.