{"record":{"id":"a9c768bd59c79075","repo":"microsoft/semantic-kernel","slug":"streaming-text-contents-not-implemented","errorCode":null,"errorMessage":"Streaming text contents not implemented.","messagePattern":"Streaming text contents not implemented\\.","errorType":"exception","errorClass":"NotImplementedError","httpStatus":null,"severity":"warning","filePath":"python/samples/concepts/auto_function_calling/nexus_raven.py","lineNumber":207,"sourceCode":"    async def _execute_function_call(\n        self, call_def: dict[str, Any], chat_history: ChatHistory, kernel: Kernel\n    ) -> FunctionResultContent:\n        \"\"\"Execute a function call.\"\"\"\n        call_def[\"fcc\"] = FunctionCallContent(\n            name=call_def[\"func\"], arguments=json.dumps(call_def[\"args\"]), id=str(call_def[\"idx\"])\n        )\n        result = await kernel.invoke_function_call(call_def[\"fcc\"], chat_history)\n        if not result:\n            call_def[\"result\"] = chat_history.messages[-1].items[0]\n        else:\n            call_def[\"result\"] = result.function_result\n\n    async def get_text_contents(self, prompt: str, settings: NexusRavenPromptExecutionSettings) -> list[TextContent]:\n        result = await self.client.text_generation(prompt, **settings.prepare_settings_dict(), stream=False)\n        return [TextContent(text=result.strip(), ai_model_id=self.ai_model_id)]\n\n    async def get_streaming_text_contents(self, prompt: str, settings: NexusRavenPromptExecutionSettings):\n        raise NotImplementedError(\"Streaming text contents not implemented.\")\n\n    def get_streaming_chat_message_contents(\n        self,\n        chat_history: \"ChatHistory\",\n        settings: \"PromptExecutionSettings\",\n        **kwargs: Any,\n    ) -> AsyncGenerator[list[\"StreamingChatMessageContent\"], Any]:\n        raise NotImplementedError(\"Streaming chat message contents not implemented.\")\n\n    def get_prompt_execution_settings_class(self) -> type[PromptExecutionSettings]:\n        return NexusRavenPromptExecutionSettings\n\n\n##########################################################\n# Step 1: Define the functions you want to articulate. ###\n##########################################################\n\n","sourceCodeStart":189,"sourceCodeEnd":225,"githubUrl":"https://github.com/microsoft/semantic-kernel/blob/c028a0c7dc4f0814cdcbaba9d998f187a41197bf/python/samples/concepts/auto_function_calling/nexus_raven.py#L189-L225","documentation":"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.","triggerScenarios":"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).","commonSituations":"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.","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."],"exampleFix":"# before\nasync def get_streaming_text_contents(self, prompt, settings):\n    raise NotImplementedError(\"Streaming text contents not implemented.\")\n# after - implement streaming over the HF client\nasync def get_streaming_text_contents(self, prompt, settings):\n    async for chunk in self.client.text_generation(\n        prompt, **settings.prepare_settings_dict(), stream=True\n    ):\n        yield [TextContent(text=chunk, ai_model_id=self.ai_model_id)]","handlingStrategy":"validation","validationCode":"def supports_streaming_text(service) -> bool:\n    import inspect\n    fn = getattr(service, \"get_streaming_text_contents\", None)\n    if fn is None:\n        return False\n    # NotImplementedError bodies are unimplemented stubs in this sample\n    try:\n        src = inspect.getsource(fn)\n    except (OSError, TypeError):\n        return True\n    return \"NotImplementedError\" not in src","typeGuard":"from typing import Protocol\nclass StreamsText(Protocol):\n    def get_streaming_text_contents(self, prompt, settings): ...\n\ndef can_stream_text(service) -> bool:\n    return supports_streaming_text(service)","tryCatchPattern":"try:\n    async for chunk in service.get_streaming_text_contents(prompt, settings):\n        ...\nexcept NotImplementedError:\n    # fall back to the non-streaming path this sample supports\n    result = await service.get_text_contents(prompt, settings)","preventionTips":["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."],"tags":["python","sample","streaming","not-implemented","nexus-raven"],"backgroundTag":null,"analyzedSha":"c028a0c7dc4f0814cdcbaba9d998f187a41197bf","analyzedAt":"2026-08-13T13:48:05.040Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}