MCP server Semantic Kernel integration

Semantic Kernel plugin model

SK organizes capabilities as plugins — named collections of functions that the LLM can discover and invoke. The KernelPlugin abstraction maps directly to MCP: one MCP server becomes one SK plugin, and each MCP tool becomes one KernelFunction.

Python: wrapping MCP tools as KernelFunctions

Each MCP tool becomes an async method on a plugin class. The @kernel_function decorator provides the name and description that SK exposes to the LLM's function calling API. The method body calls the MCP session:

import os
import asyncio
from mcp import ClientSession
from mcp.client.sse import sse_client
from semantic_kernel import Kernel
from semantic_kernel.functions import kernel_function
from semantic_kernel.connectors.ai.open_ai import AzureChatCompletion
from semantic_kernel.connectors.ai.function_choice_behavior import FunctionChoiceBehavior
from semantic_kernel.connectors.ai.open_ai import OpenAIChatPromptExecutionSettings

class SearchMCPPlugin:
    """SK plugin backed by a live MCP search server."""

    def __init__(self, session: ClientSession):
        self._session = session

    @kernel_function(
        name="search_web",
        description="Search the web for recent information on a topic. "
                    "Use for current events, recent research, or facts that may have changed since the LLM training cutoff.",
    )
    async def search_web(self, query: str, max_results: int = 5) -> str:
        """
        Args:
            query: The search query — be specific for better results
            max_results: Number of results to return (1-20)
        """
        try:
            result = await self._session.call_tool(
                "search_web",
                arguments={"query": query, "max_results": max_results},
            )
            if result.isError:
                return f"Search error: {result.content[0].text}"
            return result.content[0].text
        except Exception as e:
            return f"MCP search unavailable: {e}"

    @kernel_function(
        name="get_page_content",
        description="Retrieve and extract the main text content of a webpage. "
                    "Use after search_web to get the full content of a specific result URL.",
    )
    async def get_page_content(self, url: str) -> str:
        try:
            result = await self._session.call_tool("get_page_content", arguments={"url": url})
            if result.isError:
                return f"Fetch error: {result.content[0].text}"
            return result.content[0].text
        except Exception as e:
            return f"MCP fetch unavailable: {e}"


async def run_sk_with_mcp(question: str) -> str:
    async with sse_client(
        url=os.environ["SEARCH_MCP_URL"],
        headers={"Authorization": f"Bearer {os.environ['SEARCH_MCP_TOKEN']}"},
    ) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            kernel = Kernel()
            kernel.add_service(AzureChatCompletion(
                deployment_name=os.environ["AZURE_DEPLOYMENT"],
                endpoint=os.environ["AZURE_ENDPOINT"],
                api_key=os.environ["AZURE_API_KEY"],
            ))
            kernel.add_plugin(SearchMCPPlugin(session), plugin_name="search")

            settings = OpenAIChatPromptExecutionSettings(
                function_choice_behavior=FunctionChoiceBehavior.Auto(),
                max_tokens=2048,
            )
            from semantic_kernel.contents import ChatHistory
            history = ChatHistory()
            history.add_user_message(question)

            chat_service = kernel.get_service(type=AzureChatCompletion)
            result = await chat_service.get_chat_message_content(
                chat_history=history,
                settings=settings,
                kernel=kernel,
            )
            return str(result)

The @kernel_function description is the primary signal the LLM uses to decide when and whether to call each function. Write descriptions as explicit intent declarations: "Use when the user asks for X" reduces ambiguity more than "Returns X." Docstring Args: sections are parsed by SK into parameter descriptions that appear in the tool schema sent to the LLM.

.NET: KernelFunction attribute pattern

The .NET SK SDK uses C# attributes instead of Python decorators. The MCP .NET SDK (ModelContextProtocol) provides an IMcpClient that you inject into your plugin class:

// NuGet: Microsoft.SemanticKernel, ModelContextProtocol
using Microsoft.SemanticKernel;
using ModelContextProtocol.Client;
using System.ComponentModel;

public class SearchMcpPlugin
{
    private readonly IMcpClient _mcpClient;

    public SearchMcpPlugin(IMcpClient mcpClient)
    {
        _mcpClient = mcpClient;
    }

    [KernelFunction("search_web")]
    [Description("Search the web for recent information. Use for current events or facts " +
                 "that may have changed since the model's training cutoff.")]
    public async Task<string> SearchWebAsync(
        [Description("The search query — be specific for better results")] string query,
        [Description("Maximum results to return, 1-20")] int maxResults = 5,
        CancellationToken cancellationToken = default)
    {
        try
        {
            var result = await _mcpClient.CallToolAsync(
                "search_web",
                new Dictionary<string, object?> { ["query"] = query, ["max_results"] = maxResults },
                cancellationToken: cancellationToken);

            if (result.IsError)
                return $"Search error: {result.Content.FirstOrDefault()?.Text}";

            return result.Content.FirstOrDefault()?.Text ?? string.Empty;
        }
        catch (Exception ex)
        {
            return $"MCP search unavailable: {ex.Message}";
        }
    }
}

// Registration in DI / startup:
// var mcpClient = await McpClientFactory.CreateAsync(...);
// kernel.Plugins.AddFromObject(new SearchMcpPlugin(mcpClient), "search");

The .NET SDK's cancellation token support is a significant advantage: if the SK orchestration loop is cancelled (e.g., request timeout), the cancellation propagates into the MCP call, allowing the underlying HTTP connection to be aborted cleanly rather than hanging until server timeout.

Auto function calling with ChatCompletionAgent

SK's ChatCompletionAgent provides a higher-level abstraction: it manages the chat history, handles multi-turn function calling loops, and collects final responses. Register your MCP plugin on the agent's kernel:

from semantic_kernel.agents import ChatCompletionAgent
from semantic_kernel.contents import AuthorRole

agent = ChatCompletionAgent(
    service=AzureChatCompletion(...),
    kernel=kernel,  # kernel already has SearchMCPPlugin registered
    name="ResearchAgent",
    instructions=(
        "You are a research agent with access to a live web search MCP server. "
        "Always search for current information before answering questions about "
        "recent events, releases, or statistics. Cite your sources."
    ),
    execution_settings=OpenAIChatPromptExecutionSettings(
        function_choice_behavior=FunctionChoiceBehavior.Auto(),
    ),
)

# Multi-turn conversation
history = ChatHistory()
async for response in agent.invoke(
    ChatMessageContent(role=AuthorRole.USER, content="What MCP servers are trending this week?"),
    chat_history=history,
):
    print(response.content)

FunctionChoiceBehavior.Auto() tells SK to let the LLM decide when to call functions. Alternatives are FunctionChoiceBehavior.Required() (forces at least one function call per turn) and FunctionChoiceBehavior.NoneInvoke() (suppresses function calling, for testing). For production pipelines where you want to audit every MCP call, consider FunctionChoiceBehavior.Auto(filters={"included_plugins": ["search"]}) to restrict which plugins the LLM can invoke per turn.

Session lifecycle across multiple kernel invocations

SK kernels are often created once and reused across many requests in a web application. The MCP session — an open TCP connection to the server — must last at least as long as the kernel. The safest pattern for long-lived applications is a session factory that recreates the session on connection error:

class ResilientMCPSession:
    def __init__(self, url: str, token: str):
        self._url = url
        self._token = token
        self._session: ClientSession | None = None
        self._cm = None

    async def get_session(self) -> ClientSession:
        if self._session is None:
            await self._connect()
        return self._session

    async def _connect(self):
        if self._cm:
            try:
                await self._cm.__aexit__(None, None, None)
            except Exception:
                pass
        self._cm = sse_client(
            url=self._url,
            headers={"Authorization": f"Bearer {self._token}"},
        )
        read, write = await self._cm.__aenter__()
        self._session = ClientSession(read, write)
        await self._session.__aenter__()
        await self._session.initialize()

    async def call_tool(self, tool_name: str, arguments: dict) -> any:
        for attempt in range(3):
            try:
                session = await self.get_session()
                return await session.call_tool(tool_name, arguments=arguments)
            except Exception:
                self._session = None  # force reconnect on next call
                if attempt == 2:
                    raise
                await asyncio.sleep(0.5 * (2 ** attempt))

This pattern wraps the reconnection logic so the SK plugin class stays simple. Network partitions, server restarts, and idle-connection timeouts are handled transparently — the plugin always gets a live session without re-initialization code scattered through every kernel_function.

Monitoring MCP servers in Semantic Kernel pipelines

Enterprise SK deployments often integrate with Azure Monitor, Application Insights, or other observability stacks. MCP server health is a separate concern from SK pipeline health: the SK telemetry traces show that a KernelFunction failed, but not why the MCP server returned an error or stopped responding. Monitoring the MCP endpoint directly with AliveMCP gives you infrastructure-level visibility that complements SK-level tracing:

import logging
from semantic_kernel.filters import FunctionInvocationContext

# SK function invocation filter — logs MCP call latency
async def mcp_latency_filter(context: FunctionInvocationContext, next):
    import time
    start = time.perf_counter()
    try:
        await next(context)
    except Exception as e:
        logging.error(
            "MCP function failed",
            extra={
                "plugin": context.function.plugin_name,
                "function": context.function.name,
                "error": str(e),
                "latency_ms": (time.perf_counter() - start) * 1000,
            },
        )
        raise
    else:
        logging.info(
            "MCP function succeeded",
            extra={
                "plugin": context.function.plugin_name,
                "function": context.function.name,
                "latency_ms": (time.perf_counter() - start) * 1000,
            },
        )

kernel.add_filter("function_invocation", mcp_latency_filter)

SK filters fire on every function invocation — use them to track per-tool latency and error rates. Pair this application-level telemetry with AliveMCP uptime monitoring on your MCP endpoint so you correlate application errors ("search_web returned connection error at 14:23") with infrastructure events ("search MCP server went down at 14:22").

Further reading

• MCP server AutoGen integration — Microsoft AutoGen with MCP tools
• MCP server LangChain integration — using MCP tools in LangChain agents
• MCP server authentication — API keys, bearer tokens, and OAuth 2.0
• MCP server health check — designing a robust /health endpoint
• MCP server uptime monitoring — continuous availability checking
• AliveMCP — continuous protocol monitoring for MCP servers