Semantic Kernel setup tutorial in Python with n4n.ai

Semantic Kernel is Microsoft’s lightweight SDK for orchestrating AI plugins, planners, and memories. It works with any OpenAI-compatible endpoint, which makes it a natural fit for n4n.ai — one endpoint that reaches 240+ models with automatic fallback and per-token metering. This tutorial walks through a minimal, production-shaped setup: creating a kernel, wiring a chat service, and streaming responses. You’ll end up with a reusable foundation you can extend with planners, vector stores, or custom skills.

Prerequisites

• Python 3.10 or newer
• An n4n.ai API key (get one at n4n.ai)
• A virtual environment (recommended)

Install the Semantic Kernel package with the OpenAI connector extra:

python -m venv .venv
source .venv/bin/activate
pip install "semantic-kernel[openai]"

Verify the install:

python -c "import semantic_kernel; print(semantic_kernel.__version__)"

You should see a version string like 1.23.0 (or newer).

Configure the environment

Store your n4n.ai credentials in environment variables — never hardcode them. Create a .env file in your project root:

# .env
N4N_API_KEY=sk-...
N4N_BASE_URL=https://api.n4n.ai/v1
N4N_MODEL=gpt-4o-mini

The base URL points to the n4n.ai OpenAI-compatible endpoint. The model name can be any of the 240+ models available through the gateway; gpt-4o-mini is a fast, cost-effective default for development.

Load these values in Python using python-dotenv:

pip install python-dotenv

# config.py
import os
from dotenv import load_dotenv

load_dotenv()

N4N_API_KEY = os.getenv("N4N_API_KEY")
N4N_BASE_URL = os.getenv("N4N_BASE_URL")
N4N_MODEL = os.getenv("N4N_MODEL", "gpt-4o-mini")

if not N4N_API_KEY:
    raise RuntimeError("N4N_API_KEY not set in environment")
if not N4N_BASE_URL:
    raise RuntimeError("N4N_BASE_URL not set in environment")

Create the kernel

Semantic Kernel’s Kernel class is the central container for services, plugins, and middleware. Instantiate it once per application lifecycle (or per request in serverless contexts).

# kernel_factory.py
from semantic_kernel import Kernel
from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion
from semantic_kernel.connectors.ai.open_ai.prompt_execution_settings.open_ai_prompt_execution_settings import OpenAIChatPromptExecutionSettings

from config import N4N_API_KEY, N4N_BASE_URL, N4N_MODEL


def create_kernel() -> Kernel:
    kernel = Kernel()

    chat_service = OpenAIChatCompletion(
        ai_model_id=N4N_MODEL,
        api_key=N4N_API_KEY,
        endpoint=N4N_BASE_URL,
        service_id="n4n-chat",
    )

    kernel.add_service(chat_service)
    return kernel


def get_execution_settings() -> OpenAIChatPromptExecutionSettings:
    return OpenAIChatPromptExecutionSettings(
        service_id="n4n-chat",
        max_tokens=2048,
        temperature=0.7,
        top_p=0.95,
    )

Key points:

• service_id="n4n-chat" lets you reference this specific service when multiple are registered.
• OpenAIChatPromptExecutionSettings controls generation parameters. Adjust temperature and max_tokens for your use case.
• The endpoint parameter directs the OpenAI client to n4n.ai instead of api.openai.com.

Run a simple chat completion

With the kernel wired, invoke chat completion directly. This is the simplest path — no prompt templates, no planners, just a request and a response.

# chat_once.py
import asyncio
from semantic_kernel.contents import ChatHistory

from kernel_factory import create_kernel, get_execution_settings


async def main():
    kernel = create_kernel()
    settings = get_execution_settings()

    history = ChatHistory()
    history.add_system_message("You are a concise technical assistant.")
    history.add_user_message("Explain the difference between a kernel and a plugin in Semantic Kernel in three sentences.")

    response = await kernel.get_chat_message_content(
        chat_history=history,
        settings=settings,
    )

    print(response.content)


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

Run it:

python chat_once.py

Expected output (abridged):

A kernel is the central orchestrator that manages services, plugins, and execution pipelines. A plugin is a self-contained unit of functionality — native code or prompt templates — that the kernel can invoke. Plugins register with the kernel to expose skills like summarization, code generation, or external API calls.

The response is a ChatMessageContent object with content, role, and metadata. Access response.metadata for token usage if the provider returns it.

Stream responses for better UX

Blocking on a full completion hurts perceived latency. Semantic Kernel supports async streaming via get_streaming_chat_message_contents. This yields chunks as they arrive — ideal for CLIs, web sockets, or server-sent events.

# chat_stream.py
import asyncio
import sys
from semantic_kernel.contents import ChatHistory, StreamingChatMessageContent
from semantic_kernel.contents.utils.author_role import AuthorRole

from kernel_factory import create_kernel, get_execution_settings


async def main():
    kernel = create_kernel()
    settings = get_execution_settings()

    history = ChatHistory()
    history.add_system_message("You are a concise technical assistant.")
    history.add_user_message("Write a 5-line Python function that retries an async call with exponential backoff.")

    print("Assistant: ", end="", flush=True)

    full_response = []
    async for chunk in kernel.get_streaming_chat_message_contents(
        chat_history=history,
        settings=settings,
    ):
        # chunk is a list of StreamingChatMessageContent (one per choice)
        for message in chunk:
            if message.content:
                sys.stdout.write(message.content)
                sys.stdout.flush()
                full_response.append(message.content)

    print()  # newline after stream ends

    # Optionally persist the full response to history
    history.add_assistant_message("".join(full_response))


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

Run it:

python chat_stream.py

Output appears token-by-token:

Assistant: import asyncio
import random
from typing import Callable, TypeVar

T = TypeVar("T")

async def retry_with_backoff(
    func: Callable[[], T],
    max_attempts: int = 3,
    base_delay: float = 1.0,
) -> T:
    for attempt in range(max_attempts):
        try:
            return await func()
        except Exception:
            if attempt == max_attempts - 1:
                raise
            delay = base_delay * (2 ** attempt) + random.uniform(0, 0.1)
            await asyncio.sleep(delay)

Streaming uses the same OpenAIChatPromptExecutionSettings. The gateway honors stream: true automatically when you call the streaming method.

Add a reusable prompt template

Hardcoding prompts in Python strings doesn’t scale. Semantic Kernel supports prompt templates with {{$variable}} placeholders, stored as .skprompt files or inline strings. Here’s a template for code review:

# prompts/code_review.skprompt
system:
You are a senior engineer reviewing a pull request. Be specific, cite line numbers, and suggest concrete improvements.

user:
Review the following {{language}} code for correctness, performance, and readability:

```{{language}}
{{$code}}

Focus on:

• Bug risks
• Unnecessary complexity
• Missing error handling
• Style violations

Load and invoke it through the kernel:

```python
# prompt_template.py
import asyncio
from pathlib import Path
from semantic_kernel.contents import ChatHistory
from semantic_kernel.functions import KernelArguments

from kernel_factory import create_kernel, get_execution_settings


async def main():
    kernel = create_kernel()
    settings = get_execution_settings()

    prompt_path = Path(__file__).parent / "prompts" / "code_review.skprompt"
    prompt_template = prompt_path.read_text(encoding="utf-8")

    # Register as a semantic function
    code_review_fn = kernel.add_function(
        plugin_name="code_review",
        function_name="review",
        prompt_template=prompt_template,
        prompt_execution_settings=settings,
    )

    sample_code = """
async def fetch_user(user_id: int):
    conn = await pool.acquire()
    row = await conn.fetchrow("SELECT * FROM users WHERE id = $1", user_id)
    return row
"""

    arguments = KernelArguments(
        language="python",
        code=sample_code,
    )

    result = await kernel.invoke(code_review_fn, arguments)
    print(result)


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

Run it:

python prompt_template.py

Output (truncated):

**Bug risks**
- Line 3: `pool` is not defined in scope. Pass it as a parameter or import from a shared module.
- Line 4: No timeout on `fetchrow`. Add `timeout=5.0` to avoid hanging indefinitely.

**Unnecessary complexity**
- The function is simple enough; no over-engineering detected.

**Missing error handling**
- No try/except around the query. Wrap in try/except and convert asyncpg exceptions to domain exceptions.
- Connection not released on exception. Use `async with pool.acquire() as conn:`.

**Style violations**
- Missing type hint for return value (`asyncpg.Record | None`).
- Consider adding a docstring.

The template variables ({{language}}, {{$code}}) are supplied via KernelArguments. This separation keeps prompts version-controllable and testable.

Wire a native plugin (Python function)

Semantic Kernel shines when you mix prompt templates with native code. A native plugin is just a Python class with methods decorated by @kernel_function. Here’s a plugin that fetches a URL and returns its text — useful for RAG or browsing skills.

# plugins/web_fetch.py
import httpx
from semantic_kernel.functions import kernel_function, KernelFunctionMetadata
from semantic_kernel.functions.kernel_parameter_metadata import KernelParameterMetadata


class WebFetchPlugin:
    def __init__(self, timeout: float = 10.0):
        self._client = httpx.AsyncClient(timeout=timeout, follow_redirects=True)

    @kernel_function(
        name="fetch",
        description="Fetch a URL and return the response text",
        parameters=[
            KernelParameterMetadata(
                name="url",
                description="The URL to fetch",
                type_="str",
                is_required=True,
            ),
            KernelParameterMetadata(
                name="max_chars",
                description="Truncate response to this many characters",
                type_="int",
                is_required=False,
                default_value=8000,
            ),
        ],
    )
    async def fetch(self, url: str, max_chars: int = 8000) -> str:
        response = await self._client.get(url)
        response.raise_for_status()
        text = response.text
        if len(text) > max_chars:
            text = text[:max_chars] + f"\n... [truncated at {max_chars} chars]"
        return text

    async def close(self):
        await self._client.aclose()

Register and invoke it:

# native_plugin.py
import asyncio
from semantic_kernel.contents import ChatHistory
from semantic_kernel.functions import KernelArguments

from kernel_factory import create_kernel, get_execution_settings
from plugins.web_fetch import WebFetchPlugin


async def main():
    kernel = create_kernel()
    settings = get_execution_settings()

    web_plugin = WebFetchPlugin()
    kernel.add_plugin(web_plugin, plugin_name="web")

    # Let the model decide when to call the plugin
    history = ChatHistory()
    history.add_system_message(
        "You have access to a web fetch tool. Use it to answer questions about current content."
    )
    history.add_user_message("What's the title of the Python 3.12 release notes page on docs.python.org?")

    # First, let the model respond (it should emit a tool call)
    response = await kernel.get_chat_message_content(
        chat_history=history,
        settings=settings,
    )

    print("Model response:", response.content)

    # In a real loop, you'd parse tool calls, execute them, and feed results back.
    # For demo, call the plugin directly:
    result = await kernel.invoke(
        kernel.plugins["web"]["fetch"],
        KernelArguments(url="https://docs.python.org/3/whatsnew/3.12.html"),
    )
    print("\nPlugin result (first 500 chars):")
    print(str(result)[:500])

    await web_plugin.close()


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

Run it:

python native_plugin.py

Output:

Model response: I'll fetch the Python 3.12 release notes page to find the title.

Plugin result (first 500 chars):
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="utf-8">
<title>What's New In Python 3.12 — Python 3.12.4 documentation</title>
...

The model sees the plugin via function calling (if the underlying model supports it). For models without native tool calling, you’d implement a planner or manual loop. Semantic Kernel’s FunctionCallingExecutor handles the orchestration automatically when available.

Persist chat history

For multi-turn conversations, persist ChatHistory to a database or file. Semantic Kernel doesn’t prescribe storage — serialize the messages yourself:

# history_persist.py
import json
from pathlib import Path
from semantic_kernel.contents import ChatHistory, ChatMessageContent
from semantic_kernel.contents.utils.author_role import AuthorRole


HISTORY_PATH = Path("chat_history.json")


def save_history(history: ChatHistory) -> None:
    data = [
        {"role": msg.role.value, "content": msg.content, "name": msg.name}
        for msg in history.messages
    ]
    HISTORY_PATH.write_text(json.dumps(data, indent=2))


def load_history() -> ChatHistory:
    if not HISTORY_PATH.exists():
        return ChatHistory()
    data = json.loads(HISTORY_PATH.read_text())
    history = ChatHistory()
    for item in data:
        history.add_message(ChatMessageContent(role=AuthorRole(item["role"]), content=item["content"], name=item.get("name")))
    return history

Integrate into your chat loop:

# chat_with_history.py
import asyncio
from semantic_kernel.contents import ChatHistory

from kernel_factory import create_kernel, get_execution_settings
from history_persist import load_history, save_history


async def main():
    kernel = create_kernel()
    settings = get_execution_settings()

    history = load_history()
    if not history.messages:
        history.add_system_message("You are a helpful assistant.")

    print("Chat started. Type 'exit' to quit.")
    while True:
        user_input = input("\nYou: ").strip()
        if user_input.lower() in {"exit", "quit"}:
            break

        history.add_user_message(user_input)

        print("Assistant: ", end="", flush=True)
        full = []
        async for chunk in kernel.get_streaming_chat_message_contents(
            chat_history=history,
            settings=settings,
        ):
            for msg in chunk:
                if msg.content:
                    print(msg.content, end="", flush=True)
                    full.append(msg.content)

        print()
        history.add_assistant_message("".join(full))
        save_history(history)

    print("History saved.")


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

This gives you a working REPL with persistence across restarts.

Error handling and retries

Production code needs resilience. Wrap kernel calls with tenacity for automatic retries on transient failures (rate limits, 5xx, network blips). n4n.ai returns standard OpenAI error codes, so tenacity works out of the box.

pip install tenacity

# resilient_chat.py
import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential_jitter, retry_if_exception_type
import httpx
from semantic_kernel.contents import ChatHistory

from kernel_factory import create_kernel, get_execution_settings


@retry(
    wait=wait_exponential_jitter(initial=1, max=30),
    stop=stop_after_attempt(3),
    retry=retry_if_exception_type((httpx.HTTPStatusError, httpx.TimeoutException, ConnectionError)),
    reraise=True,
)
async def chat_with_retry(kernel, history, settings):
    return await kernel.get_chat_message_content(
        chat_history=history,
        settings=settings,
    )


async def main():
    kernel = create_kernel()
    settings = get_execution_settings()

    history = ChatHistory()
    history.add_system_message("You are a concise assistant.")
    history.add_user_message("What is the capital of France?")

    try:
        response = await chat_with_retry(kernel, history, settings)
        print(response.content)
    except Exception as e:
        print(f"Failed after retries: {e}")


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

The decorator retries up to 3 times with exponential backoff and jitter. Adjust stop_after_attempt and wait_exponential_jitter for your SLA.

Project structure recap

project/
├── .env
├── config.py
├── kernel_factory.py
├── prompts/
│   └── code_review.skprompt
├── plugins/
│   └── web_fetch.py
├── chat_once.py
├── chat_stream.py
├── prompt_template.py
├── native_plugin.py
├── history_persist.py
├── chat_with_history.py
└── resilient_chat.py

Each file is runnable independently. The kernel factory centralizes configuration so you swap models or endpoints in one place.

What’s next

• Planners: Try FunctionCallingStepwisePlanner for multi-step reasoning with native plugins.
• Vector memory: Add SemanticTextMemory with a vector store (Qdrant, Pinecone, Redis) for RAG.
• Filters: Implement IPromptRenderFilter or IFunctionInvocationFilter for logging, PII redaction, or cost tracking.
• Observability: Emit OpenTelemetry spans from kernel middleware; n4n.ai forwards provider cache-control hints you can log for cache hit analysis.
• Deployment: Package the kernel factory as a FastAPI dependency for stateless request-scoped kernels.

The foundation here — kernel creation, chat service wiring, streaming, prompt templates, native plugins, history persistence, and retry logic — covers 90% of production Semantic Kernel workloads. Extend incrementally.