MCP

Building an MCP Server for a Legacy SOAP System

Ravinder··8 min read
MCPLegacySOAPIntegrationAI
Share:
Building an MCP Server for a Legacy SOAP System

There is a class of enterprise system that refuses to die: the SOAP web service. SAP, Oracle Financials, IBM mainframe adapters, early-2000s ERP modules — they speak WSDL and XML and they will outlive most of the engineers who wrote them. When your AI agent needs to talk to one of these systems, you have two choices: generate a REST façade and maintain it forever, or build an MCP server that handles the XML translation once and exposes a clean tool surface to every LLM client.

The second option is better. Here is how to do it without modifying the legacy system at all.

Architecture Overview

The MCP server sits between the LLM agent and the SOAP endpoint. It handles WSDL parsing, XML serialization/deserialization, error normalization, and caching. The legacy system sees nothing but the same XML messages it has always seen.

graph LR subgraph AI Layer A[LLM Agent] end subgraph MCP Layer B[MCP Server] C[WSDL Parser] D[XML Builder] E[Response Mapper] F[Error Translator] end subgraph Legacy G[SOAP Endpoint] H[WSDL Definition] end A -->|tools/call JSON| B B --> C C -->|reads| H B --> D D -->|HTTP POST XML| G G -->|XML response| E E -->|structured JSON| B B -->|content JSON| A G -->|SOAP Fault| F F --> B

This architecture means the LLM agent never sees XML. It sends structured JSON arguments to the MCP tool and receives structured JSON back. All the SOAP complexity is an implementation detail of the MCP server.

Step 1: Parse the WSDL and Extract Operations

The WSDL file is the source of truth for what operations exist, what arguments they take, and what they return. Parse it once at startup and build an in-memory operation registry.

import zeep
from zeep import Client
from zeep.transports import Transport
import requests
 
def load_wsdl_client(wsdl_url: str, timeout: int = 30) -> Client:
    session = requests.Session()
    session.headers.update({
        "User-Agent": "MCP-SOAP-Bridge/1.0",
    })
    transport = Transport(session=session, timeout=timeout, operation_timeout=timeout)
    return Client(wsdl=wsdl_url, transport=transport)
 
def extract_operations(client: Client) -> dict:
    """Return a dict of operation_name -> {input_schema, output_schema}."""
    operations = {}
    for service in client.wsdl.services.values():
        for port in service.ports.values():
            for op_name, op in port.binding._operations.items():
                operations[op_name] = {
                    "input": op.input.signature(as_output=False),
                    "output": op.output.signature(as_output=True) if op.output else None,
                    "doc": op.doc or "",
                }
    return operations

Use zeep for Python — it is the most complete SOAP client available and handles WSDL imports, namespaces, and complex types reliably. The zeep client also handles WS-Security headers if your legacy system requires SOAP-level auth.

Step 2: Map WSDL Types to JSON Schema

WSDL uses XSD types. JSON Schema is what MCP tools need. Write a mapper that converts the common cases. You will not cover every XSD construct, but you will cover 90% of what enterprise SOAP services actually use.

from typing import Any
 
XSD_TO_JSON_TYPE: dict[str, dict] = {
    "string": {"type": "string"},
    "int": {"type": "integer"},
    "long": {"type": "integer"},
    "decimal": {"type": "number"},
    "boolean": {"type": "boolean"},
    "date": {"type": "string", "format": "date"},
    "dateTime": {"type": "string", "format": "date-time"},
    "base64Binary": {"type": "string", "contentEncoding": "base64"},
}
 
def xsd_element_to_json_schema(element) -> dict:
    """Recursively convert a zeep XSD element to JSON Schema."""
    xsd_type = element.type
 
    if hasattr(xsd_type, "elements"):  # complex type
        props = {}
        required = []
        for child_name, child_el in xsd_type.elements:
            props[child_name] = xsd_element_to_json_schema(child_el)
            if child_el.min_occurs > 0:
                required.append(child_name)
        schema = {"type": "object", "properties": props}
        if required:
            schema["required"] = required
        return schema
 
    # Simple type
    type_name = getattr(xsd_type, "name", "string") or "string"
    return XSD_TO_JSON_TYPE.get(type_name, {"type": "string"})

This gives you a JSON Schema you can attach to your MCP tool definition. The model gets proper type hints and the MCP SDK can validate inputs before they ever reach the SOAP call.

Step 3: Build the XML Request Dynamically

zeep handles XML serialization. You do not write raw XML. Pass a Python dict matching the operation's input structure, and zeep builds the SOAP envelope.

from zeep.helpers import serialize_object
import json
 
async def call_soap_operation(
    client: Client,
    operation_name: str,
    args: dict,
) -> dict:
    """Call a SOAP operation and return a JSON-serializable dict."""
    service = client.service
    operation_fn = getattr(service, operation_name)
 
    try:
        result = operation_fn(**args)
    except zeep.exceptions.Fault as e:
        raise SoapFaultError(
            code=e.code,
            message=e.message,
            detail=str(e.detail) if e.detail else None,
        )
    except Exception as e:
        raise SoapTransportError(str(e))
 
    # zeep returns complex objects; serialize to plain dict
    return json.loads(json.dumps(serialize_object(result), default=str))

serialize_object handles zeep's internal proxy objects and converts them to plain Python dicts. The default=str fallback handles Decimal and datetime objects that are not JSON-serializable by default.

Step 4: Error Translation

SOAP faults are not HTTP errors. A SOAP service returns HTTP 200 with a <Fault> element inside the response body. You need to translate these into MCP errors the client can reason about.

from mcp.server.fastmcp import FastMCP
from mcp import McpError
from mcp.types import ErrorData
import logging
 
class SoapFaultError(Exception):
    def __init__(self, code: str, message: str, detail: str | None):
        self.code = code
        self.message = message
        self.detail = detail
 
class SoapTransportError(Exception):
    pass
 
SOAP_FAULT_MAP = {
    "CLIENT": "Invalid request arguments sent to legacy system",
    "SERVER": "Legacy system internal error — retry may help",
    "VersionMismatch": "SOAP version mismatch — configuration error",
    "MustUnderstand": "Required SOAP header not understood",
}
 
def translate_soap_error(err: SoapFaultError) -> McpError:
    prefix = SOAP_FAULT_MAP.get(err.code, f"SOAP fault: {err.code}")
    user_message = f"{prefix}. Detail: {err.message}"
    if err.detail:
        user_message += f" ({err.detail})"
    logging.error(
        "SOAP fault",
        extra={"fault_code": err.code, "fault_string": err.message},
    )
    return McpError(ErrorData(code=-32000, message=user_message))

The translated error message is designed for the LLM: it tells the model whether the error is its fault (bad arguments) or the system's fault (retry may help), which allows the agent to make better retry decisions without hallucinating.

Step 5: Idempotency for Mutating Operations

SOAP POST operations are not idempotent by default. Network timeouts mean you may not know whether the operation succeeded. Build an idempotency layer using a request ID.

import hashlib
import redis.asyncio as redis
from datetime import timedelta
 
IDEMPOTENCY_TTL = timedelta(hours=24)
 
async def idempotent_soap_call(
    client: Client,
    operation_name: str,
    args: dict,
    idempotency_key: str,
    cache: redis.Redis,
) -> dict:
    cache_key = f"soap:idem:{idempotency_key}"
    cached = await cache.get(cache_key)
    if cached:
        return json.loads(cached)
 
    result = await call_soap_operation(client, operation_name, args)
 
    await cache.set(
        cache_key,
        json.dumps(result),
        ex=int(IDEMPOTENCY_TTL.total_seconds()),
    )
    return result

The LLM client generates the idempotency key (a UUID per distinct logical operation). If the MCP server sees the same key twice within the TTL window, it returns the cached result without calling the SOAP endpoint again. This handles the "did the order go through?" problem without requiring the legacy system to support idempotency natively.

Step 6: Wire It Up as MCP Tools

Now compose everything into MCP tool definitions. Generate one tool per SOAP operation, or curate a subset if the service is enormous.

from mcp.server.fastmcp import FastMCP
import asyncio
 
mcp = FastMCP("legacy-erp-mcp")
soap_client = load_wsdl_client(os.environ["SOAP_WSDL_URL"])
redis_client = redis.Redis.from_url(os.environ["REDIS_URL"])
 
@mcp.tool()
async def create_purchase_order(
    vendor_id: str,
    line_items: list[dict],
    delivery_date: str,
    idempotency_key: str,
) -> str:
    """
    Create a purchase order in the ERP system.
    Provide a unique idempotency_key (UUID) to safely retry on network failure.
    Returns the PO number assigned by the system.
    """
    args = {
        "VendorID": vendor_id,
        "LineItems": {"Item": line_items},
        "RequestedDeliveryDate": delivery_date,
    }
    try:
        result = await idempotent_soap_call(
            soap_client, "CreatePurchaseOrder", args,
            idempotency_key, redis_client
        )
        return f"Purchase order created: {result['PONumber']}"
    except SoapFaultError as e:
        raise translate_soap_error(e)
    except SoapTransportError as e:
        raise McpError(ErrorData(code=-32001, message=f"Transport error: {e}. Retry safe."))
 
if __name__ == "__main__":
    mcp.run()

The tool description tells the LLM that the idempotency key is the mechanism for safe retries. This is important because the model needs to understand the retry contract before it decides to retry.

Performance Considerations

SOAP services at large enterprises often have 2–5 second response times. A few optimizations that matter at scale:

Connection pooling. Configure zeep's requests.Session with HTTPAdapter(pool_connections=10, pool_maxsize=50). Default session settings leave connections on the table.

Response caching for reads. Many SOAP read operations return data that changes slowly. Cache responses in Redis with a TTL appropriate to the domain (vendor master data: 1 hour; live inventory: 0 seconds).

Async via thread pool. zeep is synchronous. Run it in a thread pool executor to avoid blocking the event loop:

import asyncio
from concurrent.futures import ThreadPoolExecutor
 
executor = ThreadPoolExecutor(max_workers=10)
 
async def async_soap_call(client, operation_name, args):
    loop = asyncio.get_event_loop()
    return await loop.run_in_executor(
        executor,
        lambda: getattr(client.service, operation_name)(**args)
    )

WSDL caching. Cache the parsed WSDL locally at startup. Do not re-fetch it on every server restart in production — legacy systems sometimes serve the WSDL from a slow endpoint.

Key Takeaways

  • The MCP server is the only component that changes — the legacy SOAP system is untouched, which means no risk of breaking existing integrations.
  • Parse the WSDL once at startup and build a type registry; use it to generate JSON Schema for tool inputs so the LLM gets proper type hints.
  • Translate SOAP faults into LLM-readable error messages that indicate whether the error is the agent's fault or the system's fault — this enables better autonomous retry decisions.
  • Build an idempotency layer in Redis keyed to a client-provided UUID; this solves the "did the mutation succeed?" problem without requiring the legacy system to support idempotency natively.
  • Run zeep (synchronous) in a thread pool executor to keep the async event loop unblocked under concurrent tool calls.
  • Cache WSDL locally and cache read operation responses in Redis — SOAP services at enterprises are often slow and this is the highest-ROI optimization you can make.
Share: