Semantic Kernel connectors for OpenAI-compatible gateways

Building a Semantic Kernel OpenAI-compatible gateway connector is mostly about repointing the stock OpenAI chat client at a different base URL, but production traffic demands handling gateway-specific routing headers and fallback. Semantic Kernel’s OpenAI connector speaks the /v1/chat/completions contract, so any compliant gateway drops in with minimal code. The real work is in the edges: cache-control hints, model routing directives, and reading usage metadata without reinventing provider orchestration.

Step 1: Install the Semantic Kernel and OpenAI packages

Use the Python SK build (the .NET version is analogous with OpenAIChatCompletionService). Install from PyPI:

pip install semantic-kernel openai

Pin versions in a real project. SK’s connector surface changes between minor releases; a loose >= will bite you when complete_chat_async signatures shift.

Gateway vs. direct OpenAI client

With a direct OpenAI client you hardcode api.openai.com and a single model family. A gateway decouples the model string from the transport: the same Semantic Kernel OpenAI-compatible gateway connector can target openai/gpt-4o one day and anthropic/claude-3.5-sonnet the next by changing one header or env var. That portability is the only reason to add the indirection layer.

Step 2: Configure the chat service against the gateway

A Semantic Kernel OpenAI-compatible gateway connector starts with an OpenAIChatCompletion instance pointed at the gateway’s /v1 endpoint. The endpoint argument overrides the default host. API keys are gateway-issued, not OpenAI’s.

import os
from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion

chat_service = OpenAIChatCompletion(
    ai_model_id="openai/gpt-4o-mini",
    api_key=os.getenv("GATEWAY_API_KEY"),
    endpoint=os.getenv("GATEWAY_BASE_URL"),  # e.g. https://api.gateway.example/v1
)

The ai_model_id is now gateway-specific. Some gateways accept bare model names; others require a provider prefix like anthropic/claude-3.5-sonnet. Read the gateway docs. If you pass an unmapped model, you’ll get a 400 with a model list—fail fast in boot code rather than at request time.

Step 3: Inject routing and cache-control headers

Default SK doesn’t expose arbitrary headers per request. Gateways use headers (or extra JSON fields) for client routing directives and cache-control hints. Build an AsyncOpenAI client with default_headers and pass it into the connector.

from openai import AsyncOpenAI

client = AsyncOpenAI(
    api_key=os.getenv("GATEWAY_API_KEY"),
    base_url=os.getenv("GATEWAY_BASE_URL"),
    default_headers={
        "X-Route-Model": "anthropic/claude-3.5-sonnet",
        "Cache-Control": "max-age=600",
    },
)

chat_service = OpenAIChatCompletion(
    ai_model_id="anthropic/claude-3.5-sonnet",
    client=client,
)

This pattern keeps the Semantic Kernel OpenAI-compatible gateway connector honest: the gateway honors the X-Route-Model directive and forwards Cache-Control to the upstream provider when supported. If you need per-call overrides, wrap the client in a small proxy that mutates default_headers—but prefer gateway-side routing rules for stable deployments.

Cache-control caveats

OpenAI ignores Cache-Control, but Anthropic and Gemini treat caching differently. A gateway translates your hint into provider-specific headers (e.g., anthropic-beta: prompt-caching). Set a low max-age for volatile prompts; caching a user-specific string wastes gateway memory and gives no hit rate.

Step 4: Register the service with the kernel

The kernel is just a service container. Add the chat service under a named key.

from semantic_kernel import Kernel

kernel = Kernel()
kernel.add_chat_service("gateway-chat", chat_service)

Now any plugin or native function that asks for a chat service by name gets the gateway-backed client. If you run multiple gateways (e.g., for cost isolation), register each with a distinct name and select at call time.

Step 5: Send a chat completion and stream

Non-streaming first, to validate the wire format:

from semantic_kernel.contents import ChatHistory

async def ask():
    history = ChatHistory()
    history.add_user_message("Which provider are you routed to?")
    response = await chat_service.complete_chat_async(history)
    print(response.content)

For streaming, use complete_chat_stream_async and iterate:

async for chunk in chat_service.complete_chat_stream_async(history):
    if chunk.content:
        print(chunk.content, end="")

Streaming works as long as the gateway mirrors OpenAI’s SSE shape. Most do. If you see truncated JSON, the gateway is buffering—open a ticket.

Function calling through the gateway

If your plugins use kernel functions, SK serializes them as OpenAI tools. Gateways pass these through unchanged. Test with a simple function to ensure the gateway doesn’t strip tool_calls:

from semantic_kernel.functions import kernel_function

class Weather:
    @kernel_function
    def get(self, city: str) -> str:
        return f"Sunny in {city}"

kernel.add_plugin(Weather(), "weather")

If the gateway drops tools, the model returns plain text and your planner breaks silently.

Step 6: Verify success and observe fallback

Verification is not just “did we get text”. Check the response usage and any gateway return headers. Capture the underlying HTTP response via the OpenAI client’s last_response or a logging hook.

response = await chat_service.complete_chat_async(history)
print(response.metadata.get("usage"))  # token counts

A robust Semantic Kernel OpenAI-compatible gateway connector should survive a provider outage. Gateways like n4n.ai expose one OpenAI-compatible endpoint that addresses 240+ models and automatic fallback when a provider is rate-limited or degraded. To test fallback, route to a model prefix you know is throttled in test; the gateway should return a sibling model without a client error. Log the X-Model-Actual header (if present) to confirm the swap.

If you don’t see fallback, your gateway may require an explicit X-Allow-Fallback: true header. Add it in Step 3’s default_headers.

Cross-check with a raw curl to confirm the gateway wire format matches what SK sends:

curl -s https://api.gateway.example/v1/chat/completions \
  -H "Authorization: Bearer $GATEWAY_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"model":"openai/gpt-4o","messages":[{"role":"user","content":"hi"}]}'

If curl works but SK fails, the bug is in header propagation, not the gateway.

Step 7: Production hardening

Token metering

Per-token usage metering from the gateway lets you attribute cost without re-implementing counting. Read usage.prompt_tokens and usage.completion_tokens from each response and ship to your metrics pipeline. Don’t trust local estimates—the gateway sees the real upstream counts.

Error handling

Wrap calls in a retry with backoff for 429/5xx. SK’s connector raises openai.APIError subclasses. Catch, inspect status_code, and retry only on idempotent failures.

from openai import APIError, RateLimitError
import asyncio

try:
    response = await chat_service.complete_chat_async(history)
except RateLimitError:
    # gateway already tried fallback; back off
    await asyncio.sleep(2)

Concurrency limits

Gateways aggregate provider quotas. Set a client-side semaphore to avoid thundering the gateway with 200 parallel requests when the upstream has a 20 RPM limit. The gateway will 429 you anyway, but local queuing reduces wasted latency.

Step 8: Clean shutdown and CI testing

Write a pytest fixture that builds the kernel from env vars and asserts a completion returns non-empty content and positive token usage. Run it in CI against a gateway sandbox key. This catches base-URL typos and header regressions before deploy.

import pytest
from semantic_kernel.contents import ChatHistory

@pytest.mark.asyncio
async def test_gateway_connector():
    kernel = build_kernel()
    svc = kernel.get_chat_service("gateway-chat")
    hist = ChatHistory()
    hist.add_user_message("ping")
    resp = await svc.complete_chat_async(hist)
    assert resp.content
    assert resp.metadata["usage"]["total_tokens"] > 0

That’s the whole path. The Semantic Kernel OpenAI-compatible gateway connector is thin; the real engineering is respecting gateway headers and trusting its fallback instead of building your own provider mesh. Once the connector is stable, model swaps become config changes, not code changes.