Microsoft Semantic Kernel consuming legacy SOAP web services (Python)

Integrating Legacy SOAP Services with Microsoft Semantic Kernel via MCP

Legacy Enterprise infrastructure relies heavily on SOAP (Simple Object Access Protocol). While robust, SOAP is verbose, XML-heavy, and difficult for modern LLMs to navigate natively compared to REST or GraphQL.

This guide provides a “Bridge Architecture” to modernize these systems. We will build a FastMCP server that acts as a translator: it accepts simple, natural-language-friendly arguments from Microsoft Semantic Kernel, executes the complex SOAP request using zeep, and returns clean JSON data to the AI agent.

The Architecture

The goal is to decouple the AI logic from the legacy protocol specifics.

Semantic Kernel (Client): The AI brain (Agent) that decides when to call the SOAP service.
MCP Server (Middleware): A Python-based FastMCP server running in Docker.
Legacy SOAP Service: The actual backend (e.g., SAP, Oracle, or generic Enterprise Bus).

graph LR
    A[Semantic Kernel Agent] -->|MCP Protocol (SSE)| B(MCP Server :8000)
    B -->|XML/SOAP| C[Legacy SOAP Endpoint]
    C -->|XML Response| B
    B -->|JSON| A

Prerequisites

Python 3.10+
Docker (for containerization)
A running SOAP service (or a WSDL file)
Libraries: fastmcp, zeep, semantic-kernel, mcp

Step 1: The MCP Server (`server.py`)

This server exposes a tool called call_soap_service. It handles the complexity of parsing the WSDL, creating the XML envelope, and managing proxies.

from fastmcp import FastMCP
from zeep import Client, Settings
from zeep.transports import Transport
import json
import requests

# Initialize FastMCP
mcp = FastMCP("SOAP-Legacy-Bridge")

@mcp.tool()
def call_soap_service(wsdl_url: str, operation_name: str, arguments_json: str = "{}") -> str:
    """
    Dynamically calls a SOAP web service operation.

    Args:
        wsdl_url (str): The URL to the WSDL definition (e.g., http://legacy-erp:8080/service?wsdl).
        operation_name (str): The specific function to call on the SOAP service.
        arguments_json (str): A JSON string dictionary of arguments to pass to the operation.
    """
    try:
        # 1. Setup Proxy (Critical for Enterprise environments)
        # For production, inject BrightData proxy URL here
        # proxies = {
        #    'http': 'http://user:[email protected]:22225',
        #    'https': 'http://user:[email protected]:22225'
        # }
        # session = requests.Session()
        # session.proxies.update(proxies)
        # transport = Transport(session=session)

        # Simple transport for demo (comment out if using proxies above)
        transport = Transport()

        # 2. strict=False allows Zeep to be forgiving with legacy non-compliant WSDLs
        settings = Settings(strict=False, xml_huge_tree=True)

        # 3. Initialize Client
        client = Client(wsdl=wsdl_url, transport=transport, settings=settings)

        # 4. Resolve the operation dynamically
        # This allows the AI to pick the operation name as a string
        service = client.service
        if not hasattr(service, operation_name):
            return json.dumps({"error": f"Operation '{operation_name}' not found in WSDL."})

        method = getattr(service, operation_name)

        # 5. Parse arguments
        try:
            args_dict = json.loads(arguments_json)
        except json.JSONDecodeError:
            return json.dumps({"error": "Invalid JSON format in arguments_json"})

        # 6. Execute Call
        # **kwargs unpacking allows flexible argument passing
        response = method(**args_dict)

        # 7. Serialize Response
        # Zeep returns custom objects, we need to convert them to standard dict/JSON
        from zeep.helpers import serialize_object
        return json.dumps(serialize_object(response), default=str)

    except Exception as e:
        return json.dumps({"error": str(e), "type": type(e).__name__})

if __name__ == "__main__":
    # HOST must be 0.0.0.0 to work inside Docker
    mcp.run(transport='sse', host='0.0.0.0', port=8000)

Step 2: The Dockerfile

We need a container that exposes port 8000 so the Semantic Kernel client can connect.

# Use an official Python runtime as a parent image
FROM python:3.11-slim

# Set the working directory
WORKDIR /app

# Install system dependencies (needed for lxml/zeep sometimes)
RUN apt-get update && apt-get install -y \
    libxml2-dev \
    libxslt-dev \
    gcc \
    && rm -rf /var/lib/apt/lists/*

# Install Python libraries
RUN pip install --no-cache-dir fastmcp zeep requests

# Copy the current directory contents into the container
COPY server.py .

# Make port 8000 available to the world outside this container
EXPOSE 8000

# Run server.py when the container launches
CMD ["python", "server.py"]

Build and Run:

docker build -t soap-mcp-bridge .
docker run -p 8000:8000 soap-mcp-bridge

Step 3: Semantic Kernel Client (`client.py`)

This client demonstrates how to consume the MCP server. We define a list mcps containing our server endpoints, and then iterate through them to register the necessary tools with Semantic Kernel.

import asyncio
import json
import os
from semantic_kernel import Kernel
from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion
from semantic_kernel.functions import kernel_function

# We use the mcp client library to talk to our server
from mcp import ClientSession, StdioServerParameters
from mcp.client.sse import sse_client

# CONFIGURATION: Define your MCP servers here
mcps = [
    "http://localhost:8000/sse"
]

class SoapMcpPlugin:
    """
    A Semantic Kernel Plugin that bridges calls to a specific MCP Server URL.
    """
    def __init__(self, mcp_url: str):
        self.mcp_url = mcp_url

    @kernel_function(
        description="Calls a legacy SOAP web service using a WSDL URL and operation name.",
        name="call_soap"
    )
    async def call_soap(self, wsdl_url: str, operation_name: str, arguments_json: str) -> str:
        """
        Connects to the MCP server via SSE and executes the SOAP request.
        """
        # Connect to the MCP Server
        async with sse_client(self.mcp_url) as streams:
            async with ClientSession(streams[0], streams[1]) as session:
                await session.initialize()

                # Execute the tool on the MCP server
                # 'call_soap_service' is the name of the function decorated with @mcp.tool() in server.py
                result = await session.call_tool(
                    "call_soap_service",
                    arguments={
                        "wsdl_url": wsdl_url,
                        "operation_name": operation_name,
                        "arguments_json": arguments_json
                    }
                )

                # Extract text content from the result
                if result.content and hasattr(result.content[0], 'text'):
                    return result.content[0].text
                return str(result)

async def main():
    # 1. Initialize Kernel
    kernel = Kernel()

    # 2. Add AI Service (OpenAI)
    # Ensure OPENAI_API_KEY is set in environment
    service_id = "default"
    kernel.add_service(
        OpenAIChatCompletion(
            service_id=service_id,
            ai_model_id="gpt-4o",
            api_key=os.getenv("OPENAI_API_KEY"),
        )
    )

    # 3. Register MCP Plugins from the configuration list
    print(f"Connecting to MCP servers: {mcps}")
    for i, url in enumerate(mcps):
        plugin_name = f"LegacySoap_Server_{i}"
        plugin = SoapMcpPlugin(mcp_url=url)
        kernel.add_plugin(plugin, plugin_name=plugin_name)

    # 4. Create a prompt that requires SOAP access
    req_settings = kernel.get_prompt_execution_settings_from_service_id(service_id)
    req_settings.function_choice_behavior = "auto"  # Allow LLM to choose the tool automatically

    # Example Prompt: Converting numbers to words using a public SOAP service
    prompt = """
    I need to convert the number 500 into words using the legacy number conversion service.
    The WSDL is: https://www.dataaccess.com/webservicesserver/NumberConversion.wso?WSDL
    The operation is: NumberToWords
    The argument key expected by the SOAP service is 'ubiNum'.

    Please execute this and tell me the result.
    """

    print("--- Sending Prompt to Semantic Kernel ---")

    # 5. Invoke the Agent
    result = await kernel.invoke_prompt(
        prompt=prompt,
        settings=req_settings
    )

    print(f"\n--- Agent Response ---\n{result}")

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

🛡️ Quality Assurance

Status: ✅ Verified
Environment: Python 3.11
Auditor: AgentRetrofit CI/CD

Transparency: This page may contain affiliate links.