{"record":{"id":"e96ee6d4d8985c28","repo":"microsoft/semantic-kernel","slug":"streaming-chat-message-contents-not-implemented","errorCode":null,"errorMessage":"Streaming chat message contents not implemented.","messagePattern":"Streaming chat message contents not implemented\\.","errorType":"exception","errorClass":"NotImplementedError","httpStatus":null,"severity":"warning","filePath":"python/samples/concepts/auto_function_calling/nexus_raven.py","lineNumber":215,"sourceCode":"        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\nclass MathPlugin:\n    @kernel_function\n    def cylinder_volume(\n        self,\n        radius: Annotated[float, \"The radius of the base of the cylinder.\"],\n        height: Annotated[float, \"The height of the cylinder.\"],\n    ):\n        \"\"\"Calculate the volume of a cylinder.\"\"\"","sourceCodeStart":197,"sourceCodeEnd":233,"githubUrl":"https://github.com/microsoft/semantic-kernel/blob/c028a0c7dc4f0814cdcbaba9d998f187a41197bf/python/samples/concepts/auto_function_calling/nexus_raven.py#L197-L233","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Use the non-streaming chat/text path for this sample.","If streaming chat is required, implement get_streaming_chat_message_contents to parse streamed tokens into StreamingChatMessageContent.","Point the streaming call at a service that natively supports streaming chat."],"exampleFix":"# before\ndef get_streaming_chat_message_contents(self, chat_history, settings, **kwargs):\n    raise NotImplementedError(\"Streaming chat message contents not implemented.\")\n# after - delegate to streaming text and wrap\nasync def get_streaming_chat_message_contents(self, chat_history, settings, **kwargs):\n    async for chunks in self.get_streaming_text_contents(\n        self._chat_to_prompt(chat_history), settings\n    ):\n        yield [\n            StreamingChatMessageContent(\n                role=AuthorRole.ASSISTANT,\n                items=[StreamingTextContent(text=c.text, choice_index=0)],\n                choice_index=0,\n                ai_model_id=self.ai_model_id,\n            )\n            for c in chunks\n        ]","handlingStrategy":"validation","validationCode":"def supports_streaming_chat(service) -> bool:\n    import inspect\n    fn = getattr(service, \"get_streaming_chat_message_contents\", None)\n    if fn is None:\n        return False\n    try:\n        return \"NotImplementedError\" not in inspect.getsource(fn)\n    except (OSError, TypeError):\n        return True","typeGuard":"from typing import Protocol\nclass StreamsChat(Protocol):\n    def get_streaming_chat_message_contents(self, chat_history, settings, **kwargs): ...\n\ndef can_stream_chat(service) -> bool:\n    return supports_streaming_chat(service)","tryCatchPattern":"try:\n    async for chunk in service.get_streaming_chat_message_contents(chat_history, settings):\n        ...\nexcept NotImplementedError:\n    content = await service.get_chat_message_content(chat_history, settings)","preventionTips":["Use the non-streaming chat path for the NexusRaven sample service.","Implement the method (parse streamed tokens into StreamingChatMessageContent) if you need streaming.","Route streaming chat to a service that natively supports it."],"tags":["python","sample","streaming","not-implemented","nexus-raven"],"backgroundTag":null,"analyzedSha":"c028a0c7dc4f0814cdcbaba9d998f187a41197bf","analyzedAt":"2026-08-13T13:48:05.040Z","schemaVersion":2},"datasetVersion":"2026-08-13T14:17:21.547Z"}