Python: Add initial MCP Connector Version

python/pyproject.toml

@@ -34,7 +34,7 @@ dependencies = [
     "numpy >= 1.25.0; python_version < '3.12'",
     "numpy >= 1.26.0; python_version >= '3.12'",
     # openai connector
-    "openai ~= 1.61",
+    "openai >= 1.61, < 1.68",
     # openapi and swagger
     "openapi_core >= 0.18,<0.20",
     "websockets >= 13, < 16",
@@ -79,6 +79,9 @@ hugging_face = [
     "sentence-transformers >= 2.2,< 4.0",
     "torch == 2.6.0"
 ]
+mcp = [
+    "mcp ~= 1.5"
+]
 mongo = [
     "pymongo >= 4.8.0, < 4.12",
     "motor >= 3.3.2,< 3.8.0"

python/samples/concepts/mcp/mcp_connector.py

@@ -0,0 +1,129 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+import asyncio
+from typing import TYPE_CHECKING
+
+from samples.concepts.setup.chat_completion_services import Services, get_chat_completion_service_and_request_settings
+from semantic_kernel import Kernel
+from semantic_kernel.connectors.ai.function_choice_behavior import FunctionChoiceBehavior
+from semantic_kernel.connectors.mcp.mcp_server_execution_settings import MCPStdioServerExecutionSettings
+from semantic_kernel.contents import ChatHistory
+from semantic_kernel.functions import KernelArguments
+
+if TYPE_CHECKING:
+    pass
+
+#####################################################################
+# This sample demonstrates how to build a conversational chatbot    #
+# using Semantic Kernel, featuring MCP Tools,                       #
+# non-streaming responses, and support for math and time plugins.   #
+# The chatbot is designed to interact with the user, call functions #
+# as needed, and return responses.                                  #
+#####################################################################
+
+# System message defining the behavior and persona of the chat bot.
+system_message = """
+You are a chat bot. Your name is Mosscap and
+you have one goal: figure out what people need.
+Your full name, should you need to know it, is
+Splendid Speckled Mosscap. You communicate
+effectively, but you tend to answer with long
+flowery prose. You are also a math wizard,
+especially for adding and subtracting.
+You also excel at joke telling, where your tone is often sarcastic.
+Once you have the answer I am looking for,
+you will return a full answer to me as soon as possible.
+"""
+
+# Create and configure the kernel.
+kernel = Kernel()
+
+# Define a chat function (a template for how to handle user input).
+chat_function = kernel.add_function(
+    prompt="{{$chat_history}}{{$user_input}}",
+    plugin_name="ChatBot",
+    function_name="Chat",
+)
+
+# You can select from the following chat completion services that support function calling:
+# - Services.OPENAI
+# - Services.AZURE_OPENAI
+# - Services.AZURE_AI_INFERENCE
+# - Services.ANTHROPIC
+# - Services.BEDROCK
+# - Services.GOOGLE_AI
+# - Services.MISTRAL_AI
+# - Services.OLLAMA
+# - Services.ONNX
+# - Services.VERTEX_AI
+# - Services.DEEPSEEK
+# Please make sure you have configured your environment correctly for the selected chat completion service.
+chat_completion_service, request_settings = get_chat_completion_service_and_request_settings(Services.AZURE_OPENAI)
+
+# Configure the function choice behavior. Here, we set it to Auto, where auto_invoke=True by default.
+# With `auto_invoke=True`, the model will automatically choose and call functions as needed.
+request_settings.function_choice_behavior = FunctionChoiceBehavior.Auto(filters={"excluded_plugins": ["ChatBot"]})
+
+kernel.add_service(chat_completion_service)
+
+# Pass the request settings to the kernel arguments.
+arguments = KernelArguments(settings=request_settings)
+
+# Create a chat history to store the system message, initial messages, and the conversation.
+history = ChatHistory()
+history.add_system_message(system_message)
+
+
+async def chat() -> bool:
+    """
+    Continuously prompt the user for input and show the assistant's response.
+    Type 'exit' to exit.
+    """
+    try:
+        user_input = input("User:> ")
+    except (KeyboardInterrupt, EOFError):
+        print("\n\nExiting chat...")
+        return False
+
+    if user_input.lower().strip() == "exit":
+        print("\n\nExiting chat...")
+        return False
+
+    arguments["user_input"] = user_input
+    arguments["chat_history"] = history
+
+    # Handle non-streaming responses
+    result = await kernel.invoke(chat_function, arguments=arguments)
+
+    # Update the chat history with the user's input and the assistant's response
+    if result:
+        print(f"Mosscap:> {result}")
+        history.add_user_message(user_input)
+        history.add_message(result.value[0])  # Capture the full context of the response
+
+    return True
+
+
+async def main() -> None:
+    # Make sure to have NPX installed and available in your PATH.
+
+    # Find the NPX executable in the system PATH.
+    import shutil
+
+    execution_settings = MCPStdioServerExecutionSettings(
+        command=shutil.which("npx"),
+        args=["-y", "@modelcontextprotocol/server-github"],
+    )
+
+    await kernel.add_plugin_from_mcp(
+        plugin_name="TestMCP",
+        execution_settings=execution_settings,
+    )
+    print("Welcome to the chat bot!\n  Type 'exit' to exit.\n")
+    chatting = True
+    while chatting:
+        chatting = await chat()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

python/semantic_kernel/connectors/mcp/__init__.py

@@ -0,0 +1,16 @@
+# Copyright (c) Microsoft. All rights reserved.
+from semantic_kernel.connectors.mcp.mcp_server_execution_settings import (
+    MCPServerExecutionSettings,
+    MCPSseServerExecutionSettings,
+    MCPStdioServerExecutionSettings,
+)
+from semantic_kernel.connectors.mcp.models.mcp_tool import MCPTool
+from semantic_kernel.connectors.mcp.models.mcp_tool_parameters import MCPToolParameters
+
+__all__ = [
+    "MCPServerExecutionSettings",
+    "MCPSseServerExecutionSettings",
+    "MCPStdioServerExecutionSettings",
+    "MCPTool",
+    "MCPToolParameters",
+]

python/semantic_kernel/connectors/mcp/mcp_manager.py

@@ -0,0 +1,64 @@
+# Copyright (c) Microsoft. All rights reserved.
+from mcp.types import ListToolsResult, Tool
+
+from semantic_kernel.connectors.mcp import (
+    MCPTool,
+    MCPToolParameters,
+)
+from semantic_kernel.connectors.mcp.mcp_server_execution_settings import (
+    MCPServerExecutionSettings,
+)
+from semantic_kernel.functions import KernelFunction, KernelFunctionFromMethod
+from semantic_kernel.functions.kernel_function_decorator import kernel_function
+from semantic_kernel.functions.kernel_parameter_metadata import KernelParameterMetadata
+from semantic_kernel.utils.feature_stage_decorator import experimental
+
+
+@experimental
+async def create_function_from_mcp_server(settings: MCPServerExecutionSettings):
+    """Loads Function from an MCP Server to KernelFunctions."""
+    async with settings.get_session() as session:
+        tools: ListToolsResult = await session.list_tools()
+        return _create_kernel_function_from_mcp_server_tools(tools, settings)
+
+
+def _create_kernel_function_from_mcp_server_tools(
+    tools: ListToolsResult, settings: MCPServerExecutionSettings
+) -> list[KernelFunction]:
+    """Loads Function from an MCP Server to KernelFunctions."""
+    return [_create_kernel_function_from_mcp_server_tool(tool, settings) for tool in tools.tools]
+
+
+def _create_kernel_function_from_mcp_server_tool(tool: Tool, settings: MCPServerExecutionSettings) -> KernelFunction:
+    """Generate a KernelFunction from a tool."""
+
+    @kernel_function(name=tool.name, description=tool.description)
+    async def mcp_tool_call(**kwargs):
+        async with settings.get_session() as session:
+            return await session.call_tool(tool.name, arguments=kwargs)
+
+    # Convert MCP Object in SK Object
+    mcp_function: MCPTool = MCPTool.from_mcp_tool(tool)
+    parameters: list[KernelParameterMetadata] = [
+        _generate_kernel_parameter_from_mcp_param(mcp_parameter) for mcp_parameter in mcp_function.parameters
+    ]
+
+    return KernelFunctionFromMethod(
+        method=mcp_tool_call,
+        parameters=parameters,
+    )
+
+
+def _generate_kernel_parameter_from_mcp_param(property: MCPToolParameters) -> KernelParameterMetadata:
+    """Generate a KernelParameterMetadata from an MCP Server."""
+    return KernelParameterMetadata(
+        name=property.name,
+        type_=property.type,
+        is_required=property.required,
+        default_value=property.default_value,
+        schema_data=property.items
+        if property.items is not None and isinstance(property.items, dict)
+        else {"type": f"{property.type}"}
+        if property.type
+        else None,
+    )

python/semantic_kernel/connectors/mcp/mcp_server_execution_settings.py

@@ -0,0 +1,74 @@
+# Copyright (c) Microsoft. All rights reserved.
+from contextlib import asynccontextmanager
+from typing import Any
+
+from mcp import ClientSession
+from mcp.client.sse import sse_client
+from mcp.client.stdio import StdioServerParameters, stdio_client
+from pydantic import Field
+
+from semantic_kernel.exceptions import KernelPluginInvalidConfigurationError
+from semantic_kernel.kernel_pydantic import KernelBaseModel
+
+
+class MCPServerExecutionSettings(KernelBaseModel):
+    """MCP server settings."""
+
+    session: ClientSession | None = None
+
+    @asynccontextmanager
+    async def get_session(self):
+        """Get or Open an MCP session."""
+        try:
+            if self.session is None:
+                # If the session is not open, create always new one
+                async with self.get_mcp_client() as (read, write), ClientSession(read, write) as session:
+                    await session.initialize()
+                    yield session
+            else:
+                # If the session is set by the user, just yield it
+                yield self.session
+        except Exception as ex:
+            raise KernelPluginInvalidConfigurationError("Failed establish MCP session.") from ex
+
+    def get_mcp_client(self):
+        """Get an MCP client."""
+        raise NotImplementedError("This method is only needed for subclasses.")
+
+
+class MCPStdioServerExecutionSettings(MCPServerExecutionSettings):
+    """MCP stdio server settings."""
+
+    command: str
+    args: list[str] = Field(default_factory=list)
+    env: dict[str, str] | None = None
+    encoding: str = "utf-8"
+
+    def get_mcp_client(self):
+        """Get an MCP stdio client."""
+        return stdio_client(
+            server=StdioServerParameters(
+                command=self.command,
+                args=self.args,
+                env=self.env,
+                encoding=self.encoding,
+            )
+        )
+
+
+class MCPSseServerExecutionSettings(MCPServerExecutionSettings):
+    """MCP sse server settings."""
+
+    url: str
+    headers: dict[str, Any] | None = None
+    timeout: float = 5
+    sse_read_timeout: float = 60 * 5
+
+    def get_mcp_client(self):
+        """Get an MCP SSE client."""
+        return sse_client(
+            url=self.url,
+            headers=self.headers,
+            timeout=self.timeout,
+            sse_read_timeout=self.sse_read_timeout,
+        )

python/semantic_kernel/connectors/mcp/models/mcp_tool.py

@@ -0,0 +1,36 @@
+# Copyright (c) Microsoft. All rights reserved.
+from mcp.types import Tool
+from pydantic import Field
+
+from semantic_kernel.connectors.mcp.models.mcp_tool_parameters import MCPToolParameters
+from semantic_kernel.exceptions import ServiceInvalidTypeError
+from semantic_kernel.kernel_pydantic import KernelBaseModel
+
+
+class MCPTool(KernelBaseModel):
+    """Semantic Kernel Class for MCP Tool."""
+
+    parameters: list[MCPToolParameters] = Field(default_factory=list)
+
+    @classmethod
+    def from_mcp_tool(cls, tool: Tool):
+        """Creates an MCPFunction instance from a tool."""
+        properties = tool.inputSchema.get("properties", None)
+        required = tool.inputSchema.get("required", None)
+        # Check if 'properties' is missing or not a dictionary
+        if properties is None or not isinstance(properties, dict):
+            raise ServiceInvalidTypeError("""Could not parse tool properties,
+            please ensure your server returns properties as a dictionary and required as an array.""")
+        if required is None or not isinstance(required, list):
+            raise ServiceInvalidTypeError("""Could not parse tool required fields,
+            please ensure your server returns required as an array.""")
+        parameters = [
+            MCPToolParameters(
+                name=prop_name,
+                required=prop_name in required,
+                **prop_details,
+            )
+            for prop_name, prop_details in properties.items()
+        ]
+
+        return cls(parameters=parameters)

python/semantic_kernel/connectors/mcp/models/mcp_tool_parameters.py

@@ -0,0 +1,12 @@
+# Copyright (c) Microsoft. All rights reserved.
+from semantic_kernel.kernel_pydantic import KernelBaseModel
+
+
+class MCPToolParameters(KernelBaseModel):
+    """Semantic Kernel Class for MCP Tool Parameters."""
+
+    name: str
+    type: str
+    required: bool = False
+    default_value: str | int | float = ""
+    items: dict | None = None