What causes "Streaming text contents not implemented." in microsoft/semantic-kernel?

Use the non-streaming path for this sample (get_text_contents / invoke, not invoke_stream).

microsoft/semantic-kernel · warning · NotImplementedError

Streaming text contents not implemented.

Error message

Streaming text contents not implemented.

What it means

NotImplementedError in the NexusRaven sample's custom service: get_streaming_text_contents is explicitly not implemented. The sample's NexusRaven client only supports non-streaming text generation (it calls client.text_generation with stream=False). Any caller requesting streaming text from this sample service hits a hard NotImplementedError.

Source

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

    async def _execute_function_call(
        self, call_def: dict[str, Any], chat_history: ChatHistory, kernel: Kernel
    ) -> FunctionResultContent:
        """Execute a function call."""
        call_def["fcc"] = FunctionCallContent(
            name=call_def["func"], arguments=json.dumps(call_def["args"]), id=str(call_def["idx"])
        )
        result = await kernel.invoke_function_call(call_def["fcc"], chat_history)
        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. ###
##########################################################

View on GitHub (pinned to c028a0c7dc)

Solutions

Use the non-streaming path for this sample (get_text_contents / invoke, not invoke_stream).
If streaming is required, implement get_streaming_text_contents by calling client.text_generation(..., stream=True) and yielding TextContent chunks.
Route the request to a service that supports streaming instead of the NexusRaven sample service.

Example fix

# before
async def get_streaming_text_contents(self, prompt, settings):
    raise NotImplementedError("Streaming text contents not implemented.")
# after - implement streaming over the HF client
async def get_streaming_text_contents(self, prompt, settings):
    async for chunk in self.client.text_generation(
        prompt, **settings.prepare_settings_dict(), stream=True
    ):
        yield [TextContent(text=chunk, ai_model_id=self.ai_model_id)]

Defensive patterns

Strategy: validation

Validate before calling

def supports_streaming_text(service) -> bool:
    import inspect
    fn = getattr(service, "get_streaming_text_contents", None)
    if fn is None:
        return False
    # NotImplementedError bodies are unimplemented stubs in this sample
    try:
        src = inspect.getsource(fn)
    except (OSError, TypeError):
        return True
    return "NotImplementedError" not in src

Type guard

from typing import Protocol
class StreamsText(Protocol):
    def get_streaming_text_contents(self, prompt, settings): ...

def can_stream_text(service) -> bool:
    return supports_streaming_text(service)

Try / catch

try:
    async for chunk in service.get_streaming_text_contents(prompt, settings):
        ...
except NotImplementedError:
    # fall back to the non-streaming path this sample supports
    result = await service.get_text_contents(prompt, settings)

Prevention

Use the non-streaming invoke path for the NexusRaven sample.
Gate streaming calls behind a capability check before invoking.
Implement the streaming method (stream=True on the HF client) if streaming is required.

When it happens

Trigger: Invoking the sample's NexusRaven service via the streaming text API path (e.g. kernel.invoke_stream on a text prompt routed to this service, or calling get_streaming_text_contents directly).

Common situations: Reusing the NexusRaven sample service and pointing generic streaming sample code at it; or wiring it into a chat loop that defaults to streaming. The underlying NexusRaven client used in the sample does not stream.

Related errors

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