Semantic Kernel n4n.ai tutorial: streaming chat responses

Streaming responses transform how users experience LLM applications. Instead of waiting seconds for a complete answer, tokens arrive incrementally, creating the perception of speed and enabling interactive patterns like typewriter effects or progressive rendering. This tutorial walks through wiring Semantic Kernel to n4n.ai for streaming chat, covering the essential configuration, the streaming API surface, and a complete runnable example.

Prerequisites

• Python 3.10 or newer
• An n4n.ai API key (get one at n4n.ai if you don’t have one)
• Basic familiarity with async Python and Semantic Kernel concepts

Install the required packages:

pip install semantic-kernel openai python-dotenv

Semantic Kernel’s OpenAI connector works with n4n.ai because n4n.ai exposes an OpenAI-compatible endpoint. No special provider package is needed.

Configure the kernel

Create a .env file in your project root:

N4N_API_KEY=sk-your-key-here
N4N_BASE_URL=https://api.n4n.ai/v1
N4N_MODEL=gpt-4o-mini

The base URL points to n4n.ai’s OpenAI-compatible endpoint. The model name can be any of the 240+ models available through the gateway — gpt-4o-mini is a good default for testing.

Now create kernel_setup.py to initialize the kernel with the n4n.ai configuration:

import os
from dotenv import load_dotenv
from semantic_kernel import Kernel
from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion

load_dotenv()

def create_kernel() -> Kernel:
    kernel = Kernel()
    
    chat_service = OpenAIChatCompletion(
        ai_model_id=os.getenv("N4N_MODEL", "gpt-4o-mini"),
        api_key=os.getenv("N4N_API_KEY"),
        endpoint=os.getenv("N4N_BASE_URL"),
    )
    
    kernel.add_service(chat_service)
    return kernel

if __name__ == "__main__":
    kernel = create_kernel()
    print(f"Kernel ready with service: {kernel.get_service(type=OpenAIChatCompletion)}")

Run it to verify connectivity:

python kernel_setup.py

Expected output:

Kernel ready with service: OpenAIChatCompletion(ai_model_id=gpt-4o-mini, ...)

Streaming with the chat completion service

Semantic Kernel exposes streaming through the get_streaming_chat_message_contents method on the chat completion service. This returns an async iterator yielding StreamingChatMessageContent chunks as they arrive.

Create streaming_chat.py:

import asyncio
import os
from dotenv import load_dotenv
from semantic_kernel import Kernel
from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion
from semantic_kernel.contents import ChatHistory, StreamingChatMessageContent
from semantic_kernel.connectors.ai.open_ai import OpenAIChatPromptExecutionSettings

load_dotenv()

async def main():
    kernel = Kernel()
    
    chat_service = OpenAIChatCompletion(
        ai_model_id=os.getenv("N4N_MODEL", "gpt-4o-mini"),
        api_key=os.getenv("N4N_API_KEY"),
        endpoint=os.getenv("N4N_BASE_URL"),
    )
    kernel.add_service(chat_service)
    
    history = ChatHistory()
    history.add_system_message("You are a concise technical assistant.")
    history.add_user_message("Explain async generators in Python in three sentences.")
    
    settings = OpenAIChatPromptExecutionSettings(
        max_tokens=200,
        temperature=0.3,
    )
    
    print("Assistant: ", end="", flush=True)
    
    async for chunk in chat_service.get_streaming_chat_message_contents(
        chat_history=history,
        settings=settings,
        kernel=kernel,
    ):
        if chunk.content:
            print(chunk.content, end="", flush=True)
    
    print()  # newline after stream completes

if __name__ == "__main__":
    asyncio.run(main())

Run it:

python streaming_chat.py

Expected output (tokens arrive incrementally):

Assistant: Async generators use 'async def' with 'yield' to produce values asynchronously. They enable lazy evaluation for I/O-bound operations like network requests. Use 'async for' to consume them in an event loop.

Each chunk is a StreamingChatMessageContent object. The content field holds the token delta. Other useful fields include role, metadata, and inner_content (the raw provider response).

Building a reusable streaming helper

Production code benefits from a small abstraction that handles history management, streaming iteration, and optional callbacks. Create streaming_helper.py:

import asyncio
from typing import AsyncIterator, Callable, Awaitable, Optional
from semantic_kernel import Kernel
from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion, OpenAIChatPromptExecutionSettings
from semantic_kernel.contents import ChatHistory, StreamingChatMessageContent

class StreamingChatClient:
    def __init__(
        self,
        kernel: Kernel,
        system_prompt: str = "You are a helpful assistant.",
        model: Optional[str] = None,
        temperature: float = 0.7,
        max_tokens: int = 1000,
    ):
        self.kernel = kernel
        self.chat_service = kernel.get_service(type=OpenAIChatCompletion)
        self.history = ChatHistory()
        self.history.add_system_message(system_prompt)
        self.settings = OpenAIChatPromptExecutionSettings(
            max_tokens=max_tokens,
            temperature=temperature,
        )
    
    async def stream(
        self,
        user_message: str,
        on_token: Optional[Callable[[str], Awaitable[None]]] = None,
    ) -> str:
        self.history.add_user_message(user_message)
        
        full_response = []
        
        async for chunk in self.chat_service.get_streaming_chat_message_contents(
            chat_history=self.history,
            settings=self.settings,
            kernel=self.kernel,
        ):
            if chunk.content:
                full_response.append(chunk.content)
                if on_token:
                    await on_token(chunk.content)
        
        response_text = "".join(full_response)
        self.history.add_assistant_message(response_text)
        return response_text
    
    def reset_history(self, system_prompt: Optional[str] = None):
        self.history = ChatHistory()
        if system_prompt:
            self.history.add_system_message(system_prompt)

Now a clean interactive loop in interactive_chat.py:

import asyncio
import os
from dotenv import load_dotenv
from semantic_kernel import Kernel
from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion
from streaming_helper import StreamingChatClient

load_dotenv()

async def main():
    kernel = Kernel()
    chat_service = OpenAIChatCompletion(
        ai_model_id=os.getenv("N4N_MODEL", "gpt-4o-mini"),
        api_key=os.getenv("N4N_API_KEY"),
        endpoint=os.getenv("N4N_BASE_URL"),
    )
    kernel.add_service(chat_service)
    
    client = StreamingChatClient(
        kernel=kernel,
        system_prompt="You are a senior Python engineer. Give practical, concise answers.",
        temperature=0.2,
        max_tokens=500,
    )
    
    print("Streaming chat ready. Type 'exit' to quit, 'reset' to clear history.\n")
    
    while True:
        user_input = input("You: ").strip()
        if user_input.lower() in ("exit", "quit"):
            break
        if user_input.lower() == "reset":
            client.reset_history()
            print("History cleared.\n")
            continue
        
        print("Assistant: ", end="", flush=True)
        
        async def print_token(token: str):
            print(token, end="", flush=True)
        
        await client.stream(user_input, on_token=print_token)
        print("\n")

if __name__ == "__main__":
    asyncio.run(main())

Run it and try a multi-turn conversation:

python interactive_chat.py

Sample session:

You: How do I handle retries with httpx?
Assistant: Use httpx.AsyncClient with a Retry transport. Configure max_attempts, backoff_factor, and status_forcelist. Wrap in a function for reuse.

You: Show me a code snippet.
Assistant: import httpx
from httpx import Retry

retry = Retry(
    total=3,
    backoff_factor=0.5,
    status_forcelist=[500, 502, 503, 504],
)

async with httpx.AsyncClient(transport=httpx.AsyncHTTPTransport(retries=retry)) as client:
    response = await client.get("https://api.example.com/data")

You: reset
History cleared.

You: What about tenacity?
Assistant: Tenacity works at a higher level — decorate any async function with @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1)). More flexible for non-HTTP retries.

Handling tool calls in streaming mode

When the model invokes tools, streaming chunks include function_call metadata. Semantic Kernel surfaces this through the metadata dictionary on StreamingChatMessageContent. Here’s how to detect and handle tool calls while streaming:

import json
from semantic_kernel.contents import StreamingChatMessageContent

async def stream_with_tools(
    chat_service: OpenAIChatCompletion,
    history: ChatHistory,
    settings: OpenAIChatPromptExecutionSettings,
    kernel: Kernel,
    available_functions: dict,
) -> str:
    full_response = []
    tool_calls_buffer = {}
    
    async for chunk in chat_service.get_streaming_chat_message_contents(
        chat_history=history,
        settings=settings,
        kernel=kernel,
    ):
        if chunk.content:
            full_response.append(chunk.content)
            print(chunk.content, end="", flush=True)
        
        # Check for function call deltas
        if chunk.metadata and "function_call" in chunk.metadata:
            fc = chunk.metadata["function_call"]
            index = fc.get("index", 0)
            
            if index not in tool_calls_buffer:
                tool_calls_buffer[index] = {"name": "", "arguments": ""}
            
            if fc.get("name"):
                tool_calls_buffer[index]["name"] = fc["name"]
            if fc.get("arguments"):
                tool_calls_buffer[index]["arguments"] += fc["arguments"]
    
    print()  # newline
    
    # Execute any accumulated tool calls
    for call_data in tool_calls_buffer.values():
        if call_data["name"] and call_data["name"] in available_functions:
            fn = available_functions[call_data["name"]]
            args = json.loads(call_data["arguments"])
            result = await fn(**args)
            print(f"[Tool {call_data['name']} returned: {result}]")
            history.add_tool_message(str(result), tool_call_id=call_data.get("id", ""))
    
    return "".join(full_response)

This pattern matters because n4n.ai routes requests across providers, and some providers stream tool calls differently. Buffering by index handles fragmented function call deltas correctly.

Error handling and fallback behavior

Network hiccups happen. Wrap the streaming loop with retry logic and leverage n4n.ai’s automatic provider fallback (which triggers on rate limits or degraded providers). Here’s a resilient wrapper:

import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
import httpx

class ResilientStreamingClient(StreamingChatClient):
    @retry(
        wait=wait_exponential(multiplier=1, min=1, max=10),
        stop=stop_after_attempt(3),
        retry=retry_if_exception_type((httpx.RequestError, httpx.HTTPStatusError)),
    )
    async def stream(self, user_message: str, on_token=None) -> str:
        try:
            return await super().stream(user_message, on_token)
        except httpx.HTTPStatusError as e:
            if e.response.status_code == 429:
                # n4n.ai handles provider fallback automatically,
                # but we still retry at the application level
                raise
            elif e.response.status_code >= 500:
                raise
            else:
                # Client errors (4xx) — don't retry
                raise RuntimeError(f"API error: {e.response.text}") from e

The @retry decorator from tenacity handles transient network errors. n4n.ai’s gateway-level fallback means a 429 from one provider often succeeds on the next attempt without code changes.

Production considerations

Token usage tracking

Each streaming chunk includes usage metadata in the final chunk. Capture it for cost monitoring:

async for chunk in chat_service.get_streaming_chat_message_contents(...):
    if chunk.metadata and "usage" in chunk.metadata:
        usage = chunk.metadata["usage"]
        print(f"\nTokens: prompt={usage.prompt_tokens}, completion={usage.completion_tokens}, total={usage.total_tokens}")

Cancellation support

Long streams should respect cancellation. Pass an asyncio.Event or CancellationToken:

async def stream_with_cancellation(
    self,
    user_message: str,
    cancel_event: asyncio.Event,
    on_token=None,
) -> str:
    self.history.add_user_message(user_message)
    full_response = []
    
    async for chunk in self.chat_service.get_streaming_chat_message_contents(...):
        if cancel_event.is_set():
            raise asyncio.CancelledError("Stream cancelled by user")
        
        if chunk.content:
            full_response.append(chunk.content)
            if on_token:
                await on_token(chunk.content)
    
    response_text = "".join(full_response)
    self.history.add_assistant_message(response_text)
    return response_text

Structured output with streaming

For JSON mode, set response_format={"type": "json_object"} in settings. The stream still delivers token-by-token — parse incrementally with a library like ijson if you need progressive validation, or buffer and parse at the end.

Complete file structure

project/
├── .env
├── kernel_setup.py
├── streaming_chat.py
├── streaming_helper.py
├── interactive_chat.py
└── requirements.txt

requirements.txt:

semantic-kernel>=1.0.0
openai>=1.0.0
python-dotenv>=1.0.0
tenacity>=8.0.0
httpx>=0.25.0

What to explore next

• Semantic Kernel planners — chain multiple streaming calls with function calling for agentic workflows
• Prompt templates — use KernelFunctionFromPrompt with streaming for reusable prompt components
• Observability — integrate OpenTelemetry to trace streaming latency per token
• Multi-modal — n4n.ai supports vision models; stream image analysis results the same way

Streaming isn’t a nice-to-have — it’s the baseline for responsive LLM interfaces. The pattern above scales from CLI tools to production APIs. Start with the helper class, add your domain logic, and ship.