Semantic Kernel Plugins: The Complete Guide to Writing Functions That Work First Time

from semantic_kernel.functions import kernel_function
from semantic_kernel.functions.kernel_parameter_metadata import KernelParameterMetadata
 
class OrderPlugin:
    """Provides functions for looking up and managing customer orders."""
 
    @kernel_function(
        name="get_order_status",
        description="Get the current status of a customer order by order ID. "
                    "Returns the status (pending, shipped, delivered, cancelled) "
                    "and estimated delivery date.",
    )
    async def get_order_status(
        self,
        order_id: str,  # type annotation required -- SK uses this for schema
    ) -> str:          # return type annotation required
        """Retrieve order status from the database."""
        # Your implementation here
        order = await db.get_order(order_id)
        return f"Order {order_id}: {order.status}, estimated delivery: {order.eta}"
 
    @kernel_function(
        name="cancel_order",
        description="Cancel a customer order. Only works for orders in 'pending' status. "
                    "Returns a confirmation message or an error if cancellation is not possible.",
    )
    async def cancel_order(
        self,
        order_id: str,
        reason: str,   # all parameters need type annotations
    ) -> str:
        result = await db.cancel_order(order_id, reason)
        msg = 'cancelled' if result.ok else 'could not be cancelled: ' + result.error
        return f"Order {order_id}: {msg}" 

# BROKEN: no type annotations -- SK cannot infer schema
@kernel_function(name="search", description="Search the knowledge base")
async def search(self, query, max_results):
    ...
 
# FIXED: explicit type annotations
@kernel_function(name="search", description="Search the knowledge base")
async def search(
    self,
    query: str,
    max_results: int = 5,  # default values are fine
) -> str:
    ...

Always annotate parameters with the most specific type possible: str, int, float, bool, list[str]. Avoid Any, Optional without a concrete type, or dict without typing. The LLM uses the schema to decide what values to pass.

# VAGUE: LLM has no idea when to call this or what to pass
@kernel_function(
    name="lookup",
    description="Look up something",
)
async def lookup(self, query: str) -> str: ...
 
# SPECIFIC: LLM knows exactly when and how to use this
@kernel_function(
    name="lookup_customer_by_email",
    description="Look up a customer account using their email address. "
                "Returns customer ID, account status, and subscription tier. "
                "Use this when the user provides or mentions an email address.",
)
async def lookup_customer_by_email(self, email: str) -> str: ...

import json
 
# RISKY: returns a dict -- SK serialises it, but the format may surprise you
@kernel_function(name="get_products", description="...")
async def get_products(self, category: str) -> dict:
    return {"products": [...], "count": 5}
 
# BETTER: return a clean, human-readable string the LLM can reason about
@kernel_function(name="get_products", description="Get products in a category. "
                "Returns a formatted list of product names and prices.")
async def get_products(self, category: str) -> str:
    products = await db.get_products(category)
    lines = [f"- {p.name}: ${p.price:.2f}" for p in products]
    return f"Found {len(products)} products in {category}:\n" + "\n".join(lines)

import asyncio
 
# BLOCKING: will stall the async event loop
@kernel_function(name="fetch_data", description="...")
def fetch_data(self, url: str) -> str:
    import requests
    return requests.get(url).text  # blocking HTTP call
 
# CORRECT: async with non-blocking HTTP
@kernel_function(name="fetch_data", description="...")
async def fetch_data(self, url: str) -> str:
    import httpx
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        return response.text

import asyncio
from semantic_kernel import Kernel
from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion
 
kernel = Kernel()
 
# Add LLM service
kernel.add_service(OpenAIChatCompletion(
    service_id="gpt4o",
    ai_model_id="gpt-4o",
    api_key="your-key",
))
 
# Register your plugin
kernel.add_plugin(OrderPlugin(), plugin_name="orders")
 
# Test a function directly (without the LLM) -- always do this first
async def test_plugin():
    result = await kernel.invoke(
        plugin_name="orders",
        function_name="get_order_status",
        order_id="ORD-12345",
    )
    print(f"Direct invocation result: {result}")
 
asyncio.run(test_plugin())

Always test plugin functions via kernel.invoke() before testing them through the agent. This verifies the function works correctly in isolation, without the complexity of LLM function selection layered on top.