Semantic Kernel planner tutorial: automatic planning

Semantic Kernel planner automatic function planning lets an LLM decompose a user goal into a sequence of function calls, then execute them without hardcoded orchestration logic. This tutorial walks through a complete, runnable implementation using the Python SDK. You will define plugins, configure the planner, execute multi-step plans, and add the guardrails that make this reliable in production.

Prerequisites

• Python 3.10+
• An OpenAI-compatible API key (OpenAI, Azure OpenAI, or a gateway like n4n.ai that exposes the same interface)
• semantic-kernel package (>= 1.0.0)

Install the dependencies:

pip install semantic-kernel python-dotenv

Create a .env file with your credentials:

# .env
OPENAI_API_KEY=sk-...
OPENAI_ORG_ID=org-...  # optional
# If using a gateway:
# OPENAI_API_BASE=https://api.n4n.ai/v1

Project structure

sk-planner-demo/
├── .env
├── main.py
├── plugins/
│   ├── __init__.py
│   ├── math_plugin.py
│   └── weather_plugin.py
└── requirements.txt

Define the plugins

Plugins are plain Python classes decorated with @kernel_function. The planner sees the function name, description, and parameter schema — so write descriptions for the model, not for yourself.

# plugins/math_plugin.py
from semantic_kernel.functions import kernel_function

class MathPlugin:
    @kernel_function(
        name="add",
        description="Add two numbers together."
    )
    def add(self, a: float, b: float) -> float:
        return a + b

    @kernel_function(
        name="multiply",
        description="Multiply two numbers together."
    )
    def multiply(self, a: float, b: float) -> float:
        return a * b

    @kernel_function(
        name="divide",
        description="Divide the first number by the second. Raises ValueError if divisor is zero."
    )
    def divide(self, a: float, b: float) -> float:
        if b == 0:
            raise ValueError("Cannot divide by zero")
        return a / b

# plugins/weather_plugin.py
from semantic_kernel.functions import kernel_function
from typing import Annotated
import random

class WeatherPlugin:
    @kernel_function(
        name="get_current_temperature",
        description="Get the current temperature in Celsius for a given city."
    )
    def get_current_temperature(
        self,
        city: Annotated[str, "The city name, e.g., 'Seattle'"]
    ) -> float:
        # Simulated — replace with a real API call
        return round(random.uniform(-5.0, 35.0), 1)

    @kernel_function(
        name="get_forecast",
        description="Get a 3-day forecast summary for a city."
    )
    def get_forecast(
        self,
        city: Annotated[str, "The city name"]
    ) -> str:
        conditions = ["sunny", "cloudy", "rainy", "snowy", "windy"]
        return ", ".join(f"Day {i}: {random.choice(conditions)}" for i in range(1, 4))

# plugins/__init__.py
from .math_plugin import MathPlugin
from .weather_plugin import WeatherPlugin

__all__ = ["MathPlugin", "WeatherPlugin"]

Configure the kernel and planner

The FunctionCallingStepwisePlanner is the current recommended planner for automatic function planning. It requests one function call at a time, observes the result, then decides the next step — this reduces hallucination compared to single-shot planners.

# main.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.planners import FunctionCallingStepwisePlanner
from semantic_kernel.planners.function_calling_stepwise_planner import (
    FunctionCallingStepwisePlannerOptions,
)
from plugins import MathPlugin, WeatherPlugin

load_dotenv()

def build_kernel() -> Kernel:
    kernel = Kernel()
    
    # Chat completion service — swap base_url for a gateway if needed
    service_id = "planner-model"
    kernel.add_service(
        OpenAIChatCompletion(
            service_id=service_id,
            ai_model_id="gpt-4o-mini",
            api_key=os.getenv("OPENAI_API_KEY"),
            org_id=os.getenv("OPENAI_ORG_ID"),
            # api_base=os.getenv("OPENAI_API_BASE"),  # uncomment for gateway
        )
    )
    
    # Register plugins
    kernel.add_plugin(MathPlugin(), plugin_name="math")
    kernel.add_plugin(WeatherPlugin(), plugin_name="weather")
    
    return kernel

async def run_planner(kernel: Kernel, goal: str) -> str:
    planner = FunctionCallingStepwisePlanner(
        service_id="planner-model",
        options=FunctionCallingStepwisePlannerOptions(
            max_iterations=10,
            max_tokens=4000,
            temperature=0.0,
        )
    )
    
    result = await planner.invoke(kernel, goal)
    return result.final_answer

async def main():
    kernel = build_kernel()
    
    # Test 1: Pure math
    print("=== Test 1: Math chain ===")
    answer = await run_planner(kernel, "What is (15 + 27) * 3 / 2?")
    print(f"Answer: {answer}\n")
    
    # Test 2: Weather + math
    print("=== Test 2: Weather + math ===")
    answer = await run_planner(
        kernel,
        "Get the current temperature in Seattle, then convert it to Fahrenheit."
    )
    print(f"Answer: {answer}\n")
    
    # Test 3: Multi-step with branching logic
    print("=== Test 3: Conditional logic ===")
    answer = await run_planner(
        kernel,
        "Get the temperature in Miami. If it's above 25°C, tell me it's beach weather. Otherwise, suggest a jacket."
    )
    print(f"Answer: {answer}")

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

Run it:

python main.py

Expected output (values will vary due to simulated weather):

=== Test 1: Math chain ===
Answer: The result of (15 + 27) * 3 / 2 is 63.0.

=== Test 2: Weather + math ===
Answer: The current temperature in Seattle is 12.3°C, which converts to 54.1°F.

=== Test 3: Conditional logic ===
Answer: The current temperature in Miami is 28.7°C. That's beach weather!

What the planner actually does

Enable debug logging to see the internal loop:

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

You will see a sequence like:

DEBUG: Calling function: math.add with args: {"a": 15, "b": 27}
DEBUG: Function result: 42
DEBUG: Calling function: math.multiply with args: {"a": 42, "b": 3}
DEBUG: Function result: 126
DEBUG: Calling function: math.divide with args: {"a": 126, "b": 2}
DEBUG: Function result: 63.0

Each iteration: the planner sends the conversation history + available functions to the model, the model responds with a tool_calls block, the planner executes the function, appends the result, and repeats until the model returns a final answer without tool calls.

Guardrails for production

Limit iterations and tokens

The FunctionCallingStepwisePlannerOptions caps prevent runaway loops:

options = FunctionCallingStepwisePlannerOptions(
    max_iterations=8,        # hard stop
    max_tokens=3000,         # context window guard
    temperature=0.0,         # deterministic planning
)

Validate function outputs

Wrap plugin methods to catch exceptions and return structured error objects the planner can reason about:

# plugins/math_plugin.py (updated divide)
from semantic_kernel.functions import kernel_function
from pydantic import BaseModel
from typing import Union

class DivisionResult(BaseModel):
    success: bool
    value: float | None = None
    error: str | None = None

class MathPlugin:
    # ... add, multiply unchanged ...
    
    @kernel_function(
        name="divide",
        description="Divide the first number by the second. Returns a result object with success flag."
    )
    def divide(self, a: float, b: float) -> DivisionResult:
        if b == 0:
            return DivisionResult(success=False, error="Division by zero")
        return DivisionResult(success=True, value=a / b)

The planner now receives a structured object and can decide to retry, ask for clarification, or abort.

Add a timeout

import asyncio

async def run_with_timeout(kernel: Kernel, goal: str, timeout_seconds: int = 30) -> str:
    try:
        return await asyncio.wait_for(run_planner(kernel, goal), timeout=timeout_seconds)
    except asyncio.TimeoutError:
        return "Planner timed out. The goal may be too complex or a function is hanging."

Restrict available functions per request

Not every user request needs every plugin. Build a kernel per request with only the relevant plugins:

def build_kernel_for_domain(domain: str) -> Kernel:
    kernel = Kernel()
    kernel.add_service(OpenAIChatCompletion(...))
    
    if domain in ("math", "general"):
        kernel.add_plugin(MathPlugin(), plugin_name="math")
    if domain in ("weather", "general"):
        kernel.add_plugin(WeatherPlugin(), plugin_name="weather")
    
    return kernel

Handling ambiguous goals

The planner struggles when the goal is underspecified. Add a clarification step before planning:

from semantic_kernel.contents import ChatHistory
from semantic_kernel.connectors.ai.open_ai import OpenAIChatPromptExecutionSettings

CLARIFICATION_PROMPT = """
The user said: "{goal}"

Available plugins: {plugin_names}

Is this request clear enough to execute? If not, ask ONE specific clarifying question.
Respond with either:
- CLEAR: <restated goal>
- CLARIFY: <question>
"""

async def clarify_goal(kernel: Kernel, goal: str, plugin_names: list[str]) -> str:
    settings = OpenAIChatPromptExecutionSettings(temperature=0.0, max_tokens=200)
    history = ChatHistory()
    history.add_user_message(CLARIFICATION_PROMPT.format(
        goal=goal,
        plugin_names=", ".join(plugin_names)
    ))
    
    result = await kernel.get_service("planner-model").get_chat_message_content(
        history, settings
    )
    return str(result).strip()

async def main_with_clarification():
    kernel = build_kernel()
    goal = "Calculate the thing."
    
    clarification = await clarify_goal(kernel, goal, ["math", "weather"])
    print(f"Clarification: {clarification}")
    
    if clarification.startswith("CLEAR:"):
        clear_goal = clarification.replace("CLEAR:", "").strip()
        answer = await run_planner(kernel, clear_goal)
        print(f"Answer: {answer}")
    else:
        print("Please clarify:", clarification.replace("CLARIFY:", "").strip())

Testing the planner in isolation

Unit test your plugins independently, then test the planner with a fake model that returns predetermined function calls. Semantic Kernel provides MockChatCompletion for this:

# test_planner.py
import pytest
from semantic_kernel import Kernel
from semantic_kernel.connectors.ai.open_ai import MockChatCompletion
from semantic_kernel.planners import FunctionCallingStepwisePlanner
from plugins import MathPlugin

@pytest.mark.asyncio
async def test_math_chain():
    kernel = Kernel()
    kernel.add_service(MockChatCompletion(
        responses=[
            # First call: add 15 + 27
            '{"tool_calls": [{"id": "1", "function": {"name": "math-add", "arguments": "{\\"a\\": 15, \\"b\\": 27}"}}]}',
            # Second call: multiply result * 3
            '{"tool_calls": [{"id": "2", "function": {"name": "math-multiply", "arguments": "{\\"a\\": 42, \\"b\\": 3}"}}]}',
            # Third call: divide result / 2
            '{"tool_calls": [{"id": "3", "function": {"name": "math-divide", "arguments": "{\\"a\\": 126, \\"b\\": 2}"}}]}',
            # Final answer
            "The result is 63.0",
        ]
    ))
    kernel.add_plugin(MathPlugin(), plugin_name="math")
    
    planner = FunctionCallingStepwisePlanner(service_id="mock")
    result = await planner.invoke(kernel, "What is (15 + 27) * 3 / 2?")
    
    assert "63" in result.final_answer

Run with pytest test_planner.py -v.

Common failure modes and fixes

When not to use automatic planning

Automatic function planning adds latency (multiple model round-trips) and non-determinism. Avoid it when:

• The workflow is fixed and known at compile time — use a deterministic chain or KernelFunctionFromPrompt instead.
• Latency budgets are tight (< 2s end-to-end).
• You need strict audit trails of every decision — the planner’s internal reasoning is opaque.

For those cases, explicitly orchestrate functions in code:

async def fixed_workflow(kernel: Kernel, a: float, b: float, c: float) -> float:
    add_fn = kernel.get_function("math", "add")
    multiply_fn = kernel.get_function("math", "multiply")
    divide_fn = kernel.get_function("math", "divide")
    
    sum_result = await kernel.invoke(add_fn, a=a, b=b)
    product_result = await kernel.invoke(multiply_fn, a=sum_result.value, b=c)
    final_result = await kernel.invoke(divide_fn, a=product_result.value, b=2.0)
    
    return final_result.value

Next steps

• Add persistent conversation memory (ChatHistory) so multi-turn planning works across requests.
• Implement a custom planner by subclassing FunctionCallingStepwisePlanner if you need domain-specific planning logic (e.g., always verify financial calculations with a second model call).
• Connect real-time streaming of intermediate steps to the UI using the planner’s callback hooks.

The complete runnable code is in the repository structure above. Start with the basic version, add guardrails incrementally, and measure latency at each step before committing to automatic planning in your critical path.