Implementing WSS-Security (mTLS) for SOAP Agents using Zeep

In the world of legacy enterprise software, security isn’t just about API keys. Many “Big Iron” SOAP services—especially in banking and insurance—rely on Mutual TLS (mTLS) and WS-Security protocols. Standard HTTP clients (and by extension, standard AI agents) will fail the handshake immediately because they don’t present a client-side certificate.

This guide provides a drop-in Model Context Protocol (MCP) server that wraps the Python zeep library. It handles the complex certificate loading and Transport Layer Security configuration, exposing a clean, agent-friendly tool for querying secure SOAP endpoints.

🏗️ Architecture

We use the FastMCP framework to create a lightweight server. The core logic uses zeep.Client with a custom requests.Session that mounts the .pem and .key files required for mTLS.

Prerequisites

Python 3.10+
Client Certificate & Key: usually provided as a .pem file (and sometimes a separate .key file).
WSDL URL: The definition file for the SOAP service.

💻 The Code

1. `server.py`

This server exposes a single tool: query_secure_soap. It handles the mTLS handshake and proxy routing internally, so the AI agent only needs to worry about the business logic (which operation to call and what data to send).

import os
import ssl
from typing import Any, Dict, Optional

from fastmcp import FastMCP
from zeep import Client, Transport
from zeep.cache import SqliteCache
from zeep.plugins import HistoryPlugin
from requests import Session

# Initialize the MCP Server
mcp = FastMCP("SecureSOAPGateway")

@mcp.tool()
def query_secure_soap(
    wsdl_url: str,
    operation_name: str,
    payload: Dict[str, Any],
    client_cert_path: str = "/app/certs/client.pem",
    client_key_path: str = "/app/certs/client.key"
) -> str:
    """
    Executes a SOAP operation against a legacy service requiring Mutual TLS (mTLS).

    Args:
        wsdl_url: The URL to the WSDL definition.
        operation_name: The specific SOAP operation to invoke (e.g., 'GetCustomerBalance').
        payload: A dictionary of arguments matching the SOAP operation signature.
        client_cert_path: Path to the public client certificate (.pem).
        client_key_path: Path to the private key (.key).

    Returns:
        Stringified result of the SOAP call or error message.
    """
    try:
        # 1. Create a Requests Session
        session = Session()

        # 2. Configure mTLS (Mutual TLS)
        # The session.cert tuple tells requests to present this cert during handshake
        if os.path.exists(client_cert_path) and os.path.exists(client_key_path):
            session.cert = (client_cert_path, client_key_path)
        else:
            return f"Error: Certificates not found at {client_cert_path} or {client_key_path}"

        # 3. Configure Proxies (Critical for Fixed IP requirements)
        # proxies = {
        #    'http': 'http://user:[email protected]:22225',
        #    'https': 'http://user:[email protected]:22225'
        # }
        # session.proxies.update(proxies) # For production, inject BrightData proxy URL here

        # 4. Disable default verify if dealing with internal self-signed CAs (Use with caution)
        # session.verify = False
        # For production, point to your internal CA bundle:
        # session.verify = "/app/certs/internal_ca_bundle.pem"

        # 5. Initialize Zeep Transport
        # We use SqliteCache to avoid re-downloading the WSDL on every request
        transport = Transport(session=session, cache=SqliteCache(path='/tmp/zeep_cache.db'))

        # 6. Create the Client
        client = Client(wsdl=wsdl_url, transport=transport)

        # 7. Dynamically resolve the operation
        service = client.service
        if not hasattr(service, operation_name):
            return f"Error: Operation '{operation_name}' not found in WSDL."

        # 8. Invoke the method
        method = getattr(service, operation_name)

        # Unpack payload into arguments
        result = method(**payload)

        # Zeep returns complex objects; convert to string/dict for the agent
        return str(result)

    except Exception as e:
        return f"SOAP Fault or Connection Error: {str(e)}"

if __name__ == "__main__":
    # Binds to 0.0.0.0 to allow external access from Docker/Agent containers
    mcp.run(transport='sse', host='0.0.0.0', port=8000)

2. `Dockerfile`

We use a slim Python image. Note the EXPOSE 8000 instruction, which is required for platforms like Railway or when networking between Docker containers.

# 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 some XML parsing libs)
RUN apt-get update && apt-get install -y \
    gcc \
    libxml2-dev \
    libxslt-dev \
    && rm -rf /var/lib/apt/lists/*

# Install Python libraries
# zeep: The SOAP client
# fastmcp[sse]: The MCP server framework
# uvicorn: ASGI server
RUN pip install --no-cache-dir \
    zeep \
    requests \
    fastmcp[sse] \
    uvicorn

# Copy the server code
COPY server.py .

# Create a directory for certificates (mount this volume at runtime)
RUN mkdir -p /app/certs

# Expose the port for the SSE stream
EXPOSE 8000

# Run the MCP server
CMD ["python", "server.py"]

🔌 Client Connection Guide (CrewAI)

How does an AI Agent (like CrewAI) talk to this mTLS-secured fortress? It simply connects to the MCP server via the mcps parameter. The agent doesn’t need to know about certificates; it just asks the tool to query the database.

CrewAI Configuration

We utilize the mcps argument in the Agent definition to directly link the agent to the running SSE stream.

from crewai import Agent, Task, Crew

# 1. Define the Agent
billing_specialist = Agent(
    role='Billing Legacy Specialist',
    goal='Retrieve invoice status from the old SOAP system',
    backstory="You are an expert at navigating the 2005-era billing mainframe.",
    # Connect directly to the SSE endpoint of our Dockerized MCP server
    mcps=["http://localhost:8000/sse"],
    verbose=True
)

# 2. Define the Task
check_invoice_task = Task(
    description="Check the status of Invoice #998877 using the 'GetInvoiceStatus' operation.",
    expected_output="The current status of the invoice.",
    agent=billing_specialist
)

# 3. Run
crew = Crew(agents=[billing_specialist], tasks=[check_invoice_task])
result = crew.kickoff()
print(result)

Running the System

Prepare Certificates: Place your client.pem and client.key in a local folder called certs/.
Build Docker:
Terminal window
```
docker build -t secure-soap-agent .
```
Run Docker: Mount your local certs folder into the container so the Python script can read them.
Terminal window
```
docker run -p 8000:8000 -v $(pwd)/certs:/app/certs secure-soap-agent
```

The server is now listening on port 8000. Your AI agent can blindly make requests, and the MCP server handles the encrypted mTLS tunnel transparently.

🛡️ Quality Assurance

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

Transparency: This page may contain affiliate links.

Implementing WSS-Security (mTLS) for SOAP Agents using Zeep

Implementing WSS-Security (mTLS) for SOAP Agents using Zeep

🏗️ Architecture

Prerequisites

💻 The Code

1. server.py

2. Dockerfile

🔌 Client Connection Guide (CrewAI)

CrewAI Configuration

Running the System

🛡️ Quality Assurance

1. `server.py`

2. `Dockerfile`