microsoft/semantic-kernel · warning · NotImplementedError

Streaming chat message contents not implemented.

Error message

Streaming chat message contents not implemented.

What it means

NotImplementedError in the NexusRaven sample service: get_streaming_chat_message_contents is not implemented. The sample service only supports non-streaming generation, so any chat-completion streaming call routed to it raises NotImplementedError. It is sample scaffolding indicating an unimplemented interface method.

Source

Thrown at python/samples/concepts/auto_function_calling/nexus_raven.py:215

        if not result:
            call_def["result"] = chat_history.messages[-1].items[0]
        else:
            call_def["result"] = result.function_result

    async def get_text_contents(self, prompt: str, settings: NexusRavenPromptExecutionSettings) -> list[TextContent]:
        result = await self.client.text_generation(prompt, **settings.prepare_settings_dict(), stream=False)
        return [TextContent(text=result.strip(), ai_model_id=self.ai_model_id)]

    async def get_streaming_text_contents(self, prompt: str, settings: NexusRavenPromptExecutionSettings):
        raise NotImplementedError("Streaming text contents not implemented.")

    def get_streaming_chat_message_contents(
        self,
        chat_history: "ChatHistory",
        settings: "PromptExecutionSettings",
        **kwargs: Any,
    ) -> AsyncGenerator[list["StreamingChatMessageContent"], Any]:
        raise NotImplementedError("Streaming chat message contents not implemented.")

    def get_prompt_execution_settings_class(self) -> type[PromptExecutionSettings]:
        return NexusRavenPromptExecutionSettings


##########################################################
# Step 1: Define the functions you want to articulate. ###
##########################################################


class MathPlugin:
    @kernel_function
    def cylinder_volume(
        self,
        radius: Annotated[float, "The radius of the base of the cylinder."],
        height: Annotated[float, "The height of the cylinder."],
    ):
        """Calculate the volume of a cylinder."""

View on GitHub (pinned to c028a0c7dc)

Solutions

  1. Use the non-streaming chat/text path for this sample.
  2. If streaming chat is required, implement get_streaming_chat_message_contents to parse streamed tokens into StreamingChatMessageContent.
  3. Point the streaming call at a service that natively supports streaming chat.

Example fix

# before
def get_streaming_chat_message_contents(self, chat_history, settings, **kwargs):
    raise NotImplementedError("Streaming chat message contents not implemented.")
# after - delegate to streaming text and wrap
async def get_streaming_chat_message_contents(self, chat_history, settings, **kwargs):
    async for chunks in self.get_streaming_text_contents(
        self._chat_to_prompt(chat_history), settings
    ):
        yield [
            StreamingChatMessageContent(
                role=AuthorRole.ASSISTANT,
                items=[StreamingTextContent(text=c.text, choice_index=0)],
                choice_index=0,
                ai_model_id=self.ai_model_id,
            )
            for c in chunks
        ]
Defensive patterns

Strategy: validation

Validate before calling

def supports_streaming_chat(service) -> bool:
    import inspect
    fn = getattr(service, "get_streaming_chat_message_contents", None)
    if fn is None:
        return False
    try:
        return "NotImplementedError" not in inspect.getsource(fn)
    except (OSError, TypeError):
        return True

Type guard

from typing import Protocol
class StreamsChat(Protocol):
    def get_streaming_chat_message_contents(self, chat_history, settings, **kwargs): ...

def can_stream_chat(service) -> bool:
    return supports_streaming_chat(service)

Try / catch

try:
    async for chunk in service.get_streaming_chat_message_contents(chat_history, settings):
        ...
except NotImplementedError:
    content = await service.get_chat_message_content(chat_history, settings)

Prevention

When it happens

Trigger: Calling the streaming chat API on the sample's NexusRaven service (e.g. chat_service.get_streaming_chat_message_contents, or a chat loop configured for streaming) routes to this unimplemented method.

Common situations: Plugging the sample service into generic chat-streaming sample code, or an auto-function-calling loop that defaults to streaming chat. The underlying NexusRaven model/client in the sample does not provide a streaming chat API.

Related errors


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