Semantic Kernel plugins tutorial: importing OpenAPI specs

Semantic Kernel’s OpenAPI plugin import lets you turn any REST API into callable functions without writing wrapper code. This tutorial walks through importing a real OpenAPI spec, handling authentication, and calling the resulting plugin functions from your kernel. You’ll end up with a pattern you can drop into production.

Prerequisites

• Python 3.10+
• An OpenAPI 3.0/3.1 spec (JSON or YAML) — we’ll use a sample Petstore spec, but you can swap in your own
• semantic-kernel and httpx installed

pip install semantic-kernel httpx pyyaml

Create a minimal OpenAPI spec

If you already have a spec, skip to the next section. Otherwise, save this as petstore.yaml — it’s a trimmed version of the classic Petstore example with just enough surface area to demonstrate the import flow.

openapi: 3.0.3
info:
  title: Petstore API
  version: 1.0.0
servers:
  - url: https://petstore3.swagger.io/api/v3
paths:
  /pets:
    get:
      operationId: listPets
      summary: List all pets
      parameters:
        - name: limit
          in: query
          schema:
            type: integer
            default: 10
      responses:
        '200':
          content:
            application/json:
              schema:
                type: array
                items:
                  $ref: '#/components/schemas/Pet'
    post:
      operationId: createPet
      summary: Create a pet
      requestBody:
        required: true
        content:
          application/json:
            schema:
              $ref: '#/components/schemas/PetInput'
      responses:
        '201':
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/Pet'
  /pets/{petId}:
    get:
      operationId: getPetById
      summary: Get a pet by ID
      parameters:
        - name: petId
          in: path
          required: true
          schema:
            type: integer
      responses:
        '200':
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/Pet'
        '404':
components:
  schemas:
    Pet:
      type: object
      properties:
        id:
          type: integer
        name:
          type: string
        tag:
          type: string
      required:
        - id
        - name
    PetInput:
      type: object
      properties:
        name:
          type: string
        tag:
          type: string
      required:
        - name

Import the spec as a plugin

Semantic Kernel provides OpenAPIFunctionExecutionParameters and ImportOpenAPIPlugin to handle the heavy lifting. The import process reads the spec, creates a function for each operation, and registers them under a plugin name you choose.

# import_plugin.py
import asyncio
import json
from pathlib import Path

from semantic_kernel import Kernel
from semantic_kernel.connectors.openapi_plugin import (
    OpenAPIFunctionExecutionParameters,
    import_openapi_plugin,
)
from semantic_kernel.functions import KernelArguments

SPEC_PATH = Path("petstore.yaml")
PLUGIN_NAME = "petstore"
BASE_URL = "https://petstore3.swagger.io/api/v3"  # overridden from spec servers


async def main() -> None:
    kernel = Kernel()

    # Import the plugin — this reads the spec and registers functions
    plugin = await import_openapi_plugin(
        kernel=kernel,
        plugin_name=PLUGIN_NAME,
        openapi_spec=SPEC_PATH.read_text(),
        execution_parameters=OpenAPIFunctionExecutionParameters(
            server_url_override=BASE_URL,
            enable_payload_namespacing=True,  # wraps request/response in namespace
        ),
    )

    print(f"Imported plugin: {plugin.name}")
    print("Available functions:")
    for func in plugin.functions.values():
        print(f"  - {func.name}: {func.description}")

    # List pets (GET /pets?limit=3)
    list_result = await kernel.invoke(
        plugin_name=PLUGIN_NAME,
        function_name="listPets",
        arguments=KernelArguments(limit=3),
    )
    print("\n--- listPets result ---")
    print(json.dumps(json.loads(str(list_result)), indent=2))

    # Create a pet (POST /pets)
    create_result = await kernel.invoke(
        plugin_name=PLUGIN_NAME,
        function_name="createPet",
        arguments=KernelArguments(
            requestBody={"name": "Kernel Kitty", "tag": "semantic"}
        ),
    )
    print("\n--- createPet result ---")
    print(json.dumps(json.loads(str(create_result)), indent=2))

    # Get the created pet by ID (assuming ID returned is 1)
    get_result = await kernel.invoke(
        plugin_name=PLUGIN_NAME,
        function_name="getPetById",
        arguments=KernelArguments(petId=1),
    )
    print("\n--- getPetById result ---")
    print(json.dumps(json.loads(str(get_result)), indent=2))


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

Run it:

python import_plugin.py

Expected output (truncated for brevity):

Imported plugin: petstore
Available functions:
  - listPets: List all pets
  - createPet: Create a pet
  - getPetById: Get a pet by ID

--- listPets result ---
[
  {"id": 1, "name": "Doggo", "tag": "dog"},
  {"id": 2, "name": "Catto", "tag": "cat"},
  {"id": 3, "name": "Birdo", "tag": "bird"}
]

--- createPet result ---
{"id": 4, "name": "Kernel Kitty", "tag": "semantic"}

--- getPetById result ---
{"id": 1, "name": "Doggo", "tag": "dog"}

Understand what the import produces

Each operation in your OpenAPI spec becomes a KernelFunction with:

• Name: the operationId from the spec (required — Semantic Kernel uses it as the function identifier)
• Parameters: derived from path, query, header, and cookie parameters plus requestBody
• Return type: str containing the raw JSON response body

The plugin is a KernelPlugin instance you can inspect, serialize, or pass to agents.

# Inspect function metadata
for name, func in plugin.functions.items():
    print(f"\n{name}:")
    print(f"  Description: {func.description}")
    print(f"  Parameters: {[p.name for p in func.parameters]}")
    print(f"  Is async: {func.is_async}")

Output:

listPets:
  Parameters: ['limit']
  Is async: True

createPet:
  Parameters: ['requestBody']
  Is async: True

getPetById:
  Parameters: ['petId']
  Is async: True

Handle authentication

Most real APIs require authentication. Semantic Kernel’s OpenAPI connector supports three patterns via OpenAPIFunctionExecutionParameters:

Bearer token (JWT, API keys in Authorization header)

from semantic_kernel.connectors.openapi_plugin import (
    OpenAPIFunctionExecutionParameters,
    AuthType,
)

execution_parameters = OpenAPIFunctionExecutionParameters(
    server_url_override=BASE_URL,
    auth_type=AuthType.BEARER,
    auth_token="eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...",  # or fetch at runtime
)

API key in header or query

execution_parameters = OpenAPIFunctionExecutionParameters(
    server_url_override=BASE_URL,
    auth_type=AuthType.API_KEY,
    api_key="sk-live-...",
    api_key_name="X-API-Key",  # header name; use `api_key_location="query"` for query param
)

Custom header injection (for mutual TLS, signatures, etc.)

If your auth doesn’t fit the above, inject headers at call time via KernelArguments:

# At invocation time
arguments = KernelArguments(
    petId=123,
    __headers={"X-Signature": "computed-hmac", "X-Timestamp": "1699999999"},
)
result = await kernel.invoke(plugin_name, "getPetById", arguments=arguments)

The __headers key is reserved — any dict passed there merges into the outgoing request headers.

Control payload namespacing

By default, enable_payload_namespacing=True wraps request bodies and unwraps response bodies under a key matching the operationId. This prevents collisions when multiple operations share parameter names.

With namespacing (default):

# Input
KernelArguments(requestBody={"name": "Fluffy"})

# Actual request body sent
{"createPet": {"name": "Fluffy"}}

# Response unwrapped from {"createPet": {...}} to just {...}

Without namespacing:

execution_parameters = OpenAPIFunctionExecutionParameters(
    server_url_override=BASE_URL,
    enable_payload_namespacing=False,
)

# Input sent as-is
KernelArguments(requestBody={"name": "Fluffy"})
# Request body: {"name": "Fluffy"}

Choose based on your API contract. Most generated OpenAPI specs expect flat bodies — disable namespacing unless the spec explicitly documents wrapped payloads.

Handle errors and retries

The OpenAPI connector surfaces HTTP errors as FunctionExecutionException with the status code and response body. Wrap invocations for resilience:

from semantic_kernel.exceptions import FunctionExecutionException
from tenacity import retry, stop_after_attempt, wait_exponential_jitter

@retry(
    wait=wait_exponential_jitter(initial=1, max=10),
    stop=stop_after_attempt(3),
)
async def safe_invoke(kernel, plugin_name, function_name, arguments):
    try:
        return await kernel.invoke(plugin_name, function_name, arguments=arguments)
    except FunctionExecutionException as e:
        # Retry on 5xx, 429; raise on 4xx
        if 500 <= e.status_code < 600 or e.status_code == 429:
            raise  # triggers tenacity retry
        raise  # non-retryable

# Usage
result = await safe_invoke(kernel, PLUGIN_NAME, "getPetById", KernelArguments(petId=999))

Use the plugin with planners and agents

Once imported, the plugin works like any native function. Planners can discover and chain its operations:

from semantic_kernel.planners import FunctionCallingStepwisePlanner
from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion

kernel.add_service(OpenAIChatCompletion(service_id="gpt-4o", ai_model_id="gpt-4o"))

planner = FunctionCallingStepwisePlanner(service_id="gpt-4o")

# Ask the planner to achieve a goal using the petstore plugin
plan = await planner.create_plan(
    goal="Find a pet named 'Kernel Kitty' and return its tag",
    kernel=kernel,
)

result = await planner.execute_plan(plan, kernel)
print(result)

The planner will call listPets, filter locally (since the API doesn’t support search), then return the tag. For production, add a searchPets operation to your spec — pushing filtering to the API is always preferable.

Validate the spec before import

Malformed specs fail at import time with opaque errors. Validate first:

from openapi_spec_validator import validate_spec
import yaml

spec_dict = yaml.safe_load(SPEC_PATH.read_text())
validate_spec(spec_dict)  # raises OpenAPIValidationError if invalid
print("Spec is valid")

Add this to your CI pipeline. A broken spec in production means broken function calling.

Production checklist

Full runnable example

Save this as complete_example.py — it includes validation, auth, error handling, and planner integration.

# complete_example.py
import asyncio
import json
import os
from pathlib import Path

import yaml
from openapi_spec_validator import validate_spec
from semantic_kernel import Kernel
from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion
from semantic_kernel.connectors.openapi_plugin import (
    OpenAPIFunctionExecutionParameters,
    AuthType,
    import_openapi_plugin,
)
from semantic_kernel.exceptions import FunctionExecutionException
from semantic_kernel.functions import KernelArguments
from semantic_kernel.planners import FunctionCallingStepwisePlanner
from tenacity import retry, stop_after_attempt, wait_exponential_jitter

SPEC_PATH = Path("petstore.yaml")
PLUGIN_NAME = "petstore"
BASE_URL = "https://petstore3.swagger.io/api/v3"
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")  # set in env


def load_and_validate_spec(path: Path) -> dict:
    spec = yaml.safe_load(path.read_text())
    validate_spec(spec)
    return spec


@retry(
    wait=wait_exponential_jitter(initial=1, max=10),
    stop=stop_after_attempt(3),
)
async def safe_invoke(kernel, plugin_name, function_name, arguments):
    try:
        return await kernel.invoke(plugin_name, function_name, arguments=arguments)
    except FunctionExecutionException as e:
        if 500 <= e.status_code < 600 or e.status_code == 429:
            raise
        raise


async def main() -> None:
    # 1. Validate spec
    spec = load_and_validate_spec(SPEC_PATH)
    print("✓ Spec validated")

    # 2. Build kernel with plugin
    kernel = Kernel()

    # Add LLM service for planner (optional)
    if OPENAI_API_KEY:
        kernel.add_service(
            OpenAIChatCompletion(service_id="gpt-4o", ai_model_id="gpt-4o", api_key=OPENAI_API_KEY)
        )

    # Import with auth config — adjust for your API
    execution_params = OpenAPIFunctionExecutionParameters(
        server_url_override=BASE_URL,
        auth_type=AuthType.NONE,  # change to BEARER/API_KEY as needed
        timeout=30.0,
        enable_payload_namespacing=False,
    )

    plugin = await import_openapi_plugin(
        kernel=kernel,
        plugin_name=PLUGIN_NAME,
        openapi_spec=json.dumps(spec),
        execution_parameters=execution_params,
    )
    print(f"✓ Plugin '{plugin.name}' imported with {len(plugin.functions)} functions")

    # 3. Direct invocation with error handling
    print("\n--- Direct invocation ---")
    pets = await safe_invoke(kernel, PLUGIN_NAME, "listPets", KernelArguments(limit=2))
    print(json.dumps(json.loads(str(pets)), indent=2))

    # 4. Planner usage (requires OPENAI_API_KEY)
    if OPENAI_API_KEY:
        print("\n--- Planner execution ---")
        planner = FunctionCallingStepwisePlanner(service_id="gpt-4o")
        plan = await planner.create_plan(
            goal="Create a pet named 'Planner Pup' with tag 'automated', then retrieve it",
            kernel=kernel,
        )
        result = await planner.execute_plan(plan, kernel)
        print(result)


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

Run it:

export OPENAI_API_KEY=sk-...  # optional, for planner demo
python complete_example.py

Next steps

• Generate specs from code: Use fastapi.openapi() or springdoc-openapi to keep specs in sync with implementation
• Compose multiple plugins: Import Stripe, GitHub, and your internal APIs into the same kernel — the planner routes across them
• Add semantic descriptions: Extend the spec with x-semantic-kernel-description on operations for better planner reasoning
• Stream large responses: For file downloads or SSE endpoints, write a native function that wraps httpx streaming — the OpenAPI connector buffers full responses

The OpenAPI import path is the fastest way to give Semantic Kernel real-world capabilities. Treat the spec as your contract, validate it in CI, and you’ll avoid the wrapper-code maintenance trap.