Semantic Kernel n4n.ai tutorial: your first chat completion

If you’re searching for a semantic kernel n4n.ai first chat completion tutorial, you want working code, not concepts. This guide takes you from a fresh environment to a streaming chat loop with function calling in about 30 minutes. We’ll use n4n.ai as the OpenAI-compatible endpoint — one URL, 240+ models, automatic fallback when a provider degrades — so you can focus on Semantic Kernel patterns instead of provider plumbing.

Prerequisites

• Python 3.10 or newer
• An n4n.ai API key (get one at n4n.ai — free tier includes generous daily tokens)
• Basic familiarity with async Python

Create a clean workspace:

mkdir sk-n4n-tutorial && cd sk-n4n-tutorial
python -m venv .venv
source .venv/bin/activate
pip install --upgrade pip
pip install semantic-kernel openai python-dotenv

Configure the connection

Semantic Kernel talks to any OpenAI-compatible endpoint through the OpenAIChatCompletion service. n4n.ai exposes that at https://api.n4n.ai/v1. Create a .env file:

# .env
N4N_API_KEY=sk-your-key-here
N4N_BASE_URL=https://api.n4n.ai/v1

Now the bootstrap script. Save as chat.py:

# 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

load_dotenv()

kernel = Kernel()

chat_service = OpenAIChatCompletion(
    ai_model_id="gpt-4o-mini",  # any model n4n.ai serves
    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, helpful assistant.")

async def main():
    print("Type 'exit' to quit.\n")
    while True:
        user = input("You: ").strip()
        if user.lower() in {"exit", "quit"}:
            break
        history.add_user_message(user)

        # Non-streaming first — easiest to verify
        response = await chat_service.get_chat_message_content(
            chat_history=history,
            settings=kernel.get_prompt_execution_settings_from_service_id("default"),
        )
        print(f"Assistant: {response}\n")
        history.add_assistant_message(str(response))

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

Run it:

python chat.py

Expected output:

Type 'exit' to quit.

You: Hello
Assistant: Hello! How can I help you today?

You: exit

If you see a response, the semantic kernel n4n.ai first chat completion path is working. The kernel located the service, authenticated through n4n.ai, and returned a completion.

Add streaming

Real apps stream. Replace the non-streaming call with get_streaming_chat_message_contents:

# chat_streaming.py
import asyncio
import os
import sys
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
from semantic_kernel.connectors.ai.open_ai import OpenAIChatPromptExecutionSettings

load_dotenv()

kernel = Kernel()

chat_service = OpenAIChatCompletion(
    ai_model_id="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, helpful assistant.")

settings = OpenAIChatPromptExecutionSettings(
    max_tokens=500,
    temperature=0.7,
    stream=True,
)

async def main():
    print("Streaming mode. Type 'exit' to quit.\n")
    while True:
        user = input("You: ").strip()
        if user.lower() in {"exit", "quit"}:
            break
        history.add_user_message(user)

        print("Assistant: ", end="", flush=True)
        full_response = []
        async for chunk in chat_service.get_streaming_chat_message_contents(
            chat_history=history,
            settings=settings,
        ):
            if chunk.content:
                print(chunk.content, end="", flush=True)
                full_response.append(chunk.content)
        print("\n")
        history.add_assistant_message("".join(full_response))

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

Run it:

python chat_streaming.py

Expected output (streaming, so characters appear progressively):

Streaming mode. Type 'exit' to quit.

You: Count to five
Assistant: One, two, three, four, five.

You: exit

Function calling — the Semantic Kernel way

Semantic Kernel shines when you attach native Python functions as “plugins” the model can invoke. Let’s add a tool that fetches the current time in a given timezone.

Create plugins/time_plugin.py:

# plugins/time_plugin.py
from datetime import datetime
from zoneinfo import ZoneInfo
from semantic_kernel.functions import kernel_function

class TimePlugin:
    @kernel_function(
        name="get_current_time",
        description="Get the current time in a specified IANA timezone",
    )
    def get_current_time(self, timezone: str) -> str:
        try:
            tz = ZoneInfo(timezone)
            now = datetime.now(tz)
            return now.strftime("%Y-%m-%d %H:%M:%S %Z%z")
        except Exception as e:
            return f"Error: {e}"

Register it in the kernel and enable function calling:

# chat_functions.py
import asyncio
import os
import sys
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
from semantic_kernel.connectors.ai.open_ai import OpenAIChatPromptExecutionSettings
from semantic_kernel.connectors.ai.function_choice_behavior import FunctionChoiceBehavior
from plugins.time_plugin import TimePlugin

load_dotenv()

kernel = Kernel()

chat_service = OpenAIChatCompletion(
    ai_model_id="gpt-4o-mini",
    api_key=os.getenv("N4N_API_KEY"),
    endpoint=os.getenv("N4N_BASE_URL"),
)
kernel.add_service(chat_service)

# Register plugin
kernel.add_plugin(TimePlugin(), plugin_name="time")

history = ChatHistory()
history.add_system_message(
    "You are a helpful assistant. Use the time plugin when users ask for the time."
)

settings = OpenAIChatPromptExecutionSettings(
    max_tokens=500,
    temperature=0.7,
    stream=True,
    function_choice_behavior=FunctionChoiceBehavior.Auto(),
)

async def main():
    print("Function calling enabled. Ask for the time in any timezone. Type 'exit' to quit.\n")
    while True:
        user = input("You: ").strip()
        if user.lower() in {"exit", "quit"}:
            break
        history.add_user_message(user)

        print("Assistant: ", end="", flush=True)
        full_response = []
        async for chunk in chat_service.get_streaming_chat_message_contents(
            chat_history=history,
            settings=settings,
            kernel=kernel,  # pass kernel so function calls resolve
        ):
            if chunk.content:
                print(chunk.content, end="", flush=True)
                full_response.append(chunk.content)
        print("\n")
        history.add_assistant_message("".join(full_response))

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

Run it:

python chat_functions.py

Test the function call:

Function calling enabled. Ask for the time in any timezone. Type 'exit' to quit.

You: What time is it in Tokyo?
Assistant: The current time in Tokyo is 2025-01-15 03:42:11 JST+0900.

You: And in New York?
Assistant: The current time in New York is 2025-01-14 13:42:11 EST-0500.

You: exit

The model invoked time.get_current_time twice — once for Asia/Tokyo, once for America/New_York — and synthesized the answers. You didn’t parse tool calls manually; Semantic Kernel handled the round trip.

Structured output with Pydantic

For production, you’ll want typed responses. Semantic Kernel supports OpenAIChatPromptExecutionSettings with response_format pointing at a Pydantic model.

# structured.py
import asyncio
import os
from dotenv import load_dotenv
from pydantic import BaseModel, Field
from semantic_kernel import Kernel
from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion
from semantic_kernel.contents import ChatHistory
from semantic_kernel.connectors.ai.open_ai import OpenAIChatPromptExecutionSettings

load_dotenv()

class TaskPlan(BaseModel):
    title: str = Field(description="Short title for the plan")
    steps: list[str] = Field(description="Ordered steps to accomplish the goal")
    estimated_minutes: int = Field(description="Total estimated time in minutes")

kernel = Kernel()
chat_service = OpenAIChatCompletion(
    ai_model_id="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 planning assistant. Output only valid JSON matching the schema.")

settings = OpenAIChatPromptExecutionSettings(
    max_tokens=500,
    temperature=0.3,
    response_format=TaskPlan,
)

async def main():
    user = "Plan a 30-minute workout for a beginner at home with no equipment"
    history.add_user_message(user)

    response = await chat_service.get_chat_message_content(
        chat_history=history,
        settings=settings,
    )
    plan = TaskPlan.model_validate_json(str(response))
    print(f"Title: {plan.title}")
    print(f"Estimated: {plan.estimated_minutes} minutes")
    for i, step in enumerate(plan.steps, 1):
        print(f"  {i}. {step}")

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

Output:

Title: Beginner 30-Minute No-Equipment Home Workout
Estimated: 30 minutes
  1. Warm up: March in place for 2 minutes, arm circles 30 seconds each direction
  2. Bodyweight squats: 3 sets of 12 reps, rest 45 seconds between sets
  3. Push-ups (knees okay): 3 sets of 8-10 reps, rest 45 seconds
  4. Glute bridges: 3 sets of 15 reps, rest 30 seconds
  5. Plank: 3 sets of 30 seconds, rest 30 seconds
  6. Cool down: Stretch hamstrings, chest, and shoulders for 3 minutes

Handling provider fallback

n4n.ai routes your request across multiple upstream providers. If one degrades or rate-limits, the gateway retries another automatically. You see this as a brief latency spike, not an exception. To observe it, you can enable debug logging:

import logging
logging.basicConfig(level=logging.DEBUG)

In the logs you’ll see the x-n4n-provider header on responses indicating which upstream served the request. When a fallback occurs, the header changes. Your code doesn’t change — the semantic kernel n4n.ai first chat completion path stays the same.

Common pitfalls

Wrong model ID — Use a model n4n.ai actually serves. Run curl -H "Authorization: Bearer $N4N_API_KEY" https://api.n4n.ai/v1/models to list them. gpt-4o-mini, claude-3-5-sonnet, llama-3.1-70b are safe defaults.

Missing kernel= in streaming — Function calling only works when you pass the kernel instance to get_streaming_chat_message_contents. Without it, the model emits tool calls as text instead of executing them.

Blocking the event loop — All Semantic Kernel I/O is async. Don’t call .result() or asyncio.run() inside an already-running loop. Structure your entry point as a single asyncio.run(main()).

Token limits — n4n.ai honors each provider’s context window. If you hit a 400 error about token count, reduce max_tokens or summarize history before sending.

What’s next

You now have a working semantic kernel n4n.ai first chat completion stack with streaming, function calling, and structured output. From here:

• Add a VectorStoreMemory plugin for RAG over your documents
• Build a multi-agent orchestration with GroupChat and Agent classes
• Wrap the kernel in a FastAPI endpoint for a production service
• Enable n4n.ai’s per-token usage metering by reading x-n4n-usage response headers for cost tracking

The kernel abstraction means swapping models or providers later is a one-line config change — no refactor of your plugin logic. That’s the point.