Part 3: Building SOAP Services with Python and spyne

Why Build a SOAP Server?

Most of the time, developers interact with SOAP as a client — consuming an existing enterprise service. But there are situations where you need to expose your own service using SOAP:

• You are building an API that must integrate with an enterprise platform that only speaks SOAP (SAP, Oracle, legacy CRM)

• You are standing up a mock service to test your client code independently

• You are migrating an old Java or .NET SOAP service to Python and need to maintain the same contract

I built my first spyne service when I needed to provide a data feed to an insurance platform that required a specific WSDL contract. The platform could not consume REST. This section uses a similar inventory management service as the example — the kind of service you would genuinely build for a B2B integration.

spyne Overview

spyne is a Python library for building SOAP (and other protocol) services. It handles:

• WSDL generation based on your Python classes

• XML serialisation and deserialisation

• SOAP Fault generation

• Multiple transports (HTTP via WSGI, TCP, etc.)

The core concepts in spyne:

Concept	Purpose
ServiceBase	Base class for your service; define @rpc methods here
@rpc	Decorator that marks a method as a SOAP operation
ComplexModel	Base class for defining XSD complex types
Application	Wires the service, protocol, and transport together
WsgiApplication	WSGI adapter for hosting over HTTP

Project Structure

soap_inventory/
├── models.py           # ComplexModel type definitions
├── service.py          # ServiceBase with @rpc operations
├── app.py              # WSGI application setup
├── server.py           # Flask server entrypoint
└── requirements.txt

Step 1: Define Data Models

Define your XSD types as Python classes that extend ComplexModel.

# soap_inventory/models.py
from spyne import ComplexModel, String, Decimal, Boolean, Integer, Array
from spyne.model import Unicode


class ProductInput(ComplexModel):
    """Input model for creating or updating a product."""
    class Attributes(ComplexModel.Attributes):
        # The XSD namespace for these types
        target_namespace = "http://example.com/inventory"

    ProductId = String(min_occurs=1, max_occurs=1)
    Name = String(min_occurs=1, max_occurs=1)
    Price = Decimal(min_occurs=1, max_occurs=1)
    InStock = Boolean(min_occurs=0, max_occurs=1)


class Product(ComplexModel):
    """Full product representation returned in responses."""
    class Attributes(ComplexModel.Attributes):
        target_namespace = "http://example.com/inventory"

    ProductId = String
    Name = String
    Price = Decimal
    InStock = Boolean
    StockCount = Integer


class ProductList(ComplexModel):
    """A paginated list of products."""
    class Attributes(ComplexModel.Attributes):
        target_namespace = "http://example.com/inventory"

    Products = Array(Product)
    TotalCount = Integer
    Page = Integer
    PageSize = Integer


class StockUpdateResult(ComplexModel):
    """Result of a stock count update operation."""
    class Attributes(ComplexModel.Attributes):
        target_namespace = "http://example.com/inventory"

    ProductId = String
    OldStockCount = Integer
    NewStockCount = Integer
    UpdatedAt = String  # ISO 8601 string; use DateTime for native datetime handling

ComplexModel classes map directly to XSD complexType definitions in the generated WSDL. When spyne generates the WSDL, it reads these class attributes and produces the corresponding XSD automatically.

Step 2: Define the Service

The service class inherits from ServiceBase and defines operations using the @rpc decorator.

# soap_inventory/service.py
from datetime import datetime, timezone
from decimal import Decimal

from spyne import ServiceBase, rpc, String, Integer, Unicode
from spyne.error import ResourceNotFoundError, ArgumentError

from .models import Product, ProductInput, ProductList, StockUpdateResult

# In-memory store for this example.
# In a real project, replace this with a database query.
_PRODUCTS: dict[str, Product] = {
    "SKU-001": Product(
        ProductId="SKU-001",
        Name="Widget Pro",
        Price=Decimal("29.99"),
        InStock=True,
        StockCount=150,
    ),
    "SKU-002": Product(
        ProductId="SKU-002",
        Name="Gadget Lite",
        Price=Decimal("14.50"),
        InStock=True,
        StockCount=42,
    ),
}


class InventoryService(ServiceBase):
    """
    SOAP service for product inventory management.
    Intended for B2B integration with partner systems.
    """

    @rpc(String, _returns=Product)
    def GetProduct(ctx, product_id: str) -> Product:
        """
        Retrieve a single product by its ID.

        Args:
            product_id: The unique product SKU identifier.

        Returns:
            Product: The full product record.

        Raises:
            ResourceNotFoundError: When no product exists with the given ID.
        """
        product = _PRODUCTS.get(product_id)
        if product is None:
            raise ResourceNotFoundError(product_id)
        return product

    @rpc(Integer, Integer, _returns=ProductList)
    def ListProducts(ctx, page: int, page_size: int) -> ProductList:
        """
        Return a paginated list of all products.

        Args:
            page: Page number (1-based).
            page_size: Number of products per page.

        Returns:
            ProductList: Paginated product list with total count.
        """
        if page_size is None or page_size <= 0:
            page_size = 10
        if page is None or page <= 0:
            page = 1

        all_products = list(_PRODUCTS.values())
        start = (page - 1) * page_size
        end = start + page_size
        page_items = all_products[start:end]

        return ProductList(
            Products=page_items,
            TotalCount=len(all_products),
            Page=page,
            PageSize=page_size,
        )

    @rpc(ProductInput, _returns=Product)
    def CreateProduct(ctx, product_input: ProductInput) -> Product:
        """
        Create a new product.

        Args:
            product_input: Product data including ID, name, and price.

        Returns:
            Product: The newly created product record.

        Raises:
            ArgumentError: When a product with the given ID already exists.
        """
        if product_input.ProductId in _PRODUCTS:
            raise ArgumentError(
                f"Product already exists: {product_input.ProductId}"
            )

        new_product = Product(
            ProductId=product_input.ProductId,
            Name=product_input.Name,
            Price=product_input.Price,
            InStock=product_input.InStock if product_input.InStock is not None else True,
            StockCount=0,
        )
        _PRODUCTS[new_product.ProductId] = new_product
        return new_product

    @rpc(String, Integer, _returns=StockUpdateResult)
    def UpdateStock(ctx, product_id: str, new_stock_count: int) -> StockUpdateResult:
        """
        Update the stock count for a product.

        Args:
            product_id: The SKU of the product to update.
            new_stock_count: The new stock quantity (must be >= 0).

        Returns:
            StockUpdateResult: Before/after stock levels and timestamp.

        Raises:
            ResourceNotFoundError: When product does not exist.
            ArgumentError: When new_stock_count is negative.
        """
        if new_stock_count < 0:
            raise ArgumentError("Stock count cannot be negative")

        product = _PRODUCTS.get(product_id)
        if product is None:
            raise ResourceNotFoundError(product_id)

        old_count = product.StockCount
        product.StockCount = new_stock_count
        product.InStock = new_stock_count > 0

        return StockUpdateResult(
            ProductId=product_id,
            OldStockCount=old_count,
            NewStockCount=new_stock_count,
            UpdatedAt=datetime.now(timezone.utc).isoformat(),
        )

Understanding @rpc

The @rpc decorator is the core of spyne's operation definition.

@rpc(String, Integer, _returns=StockUpdateResult)
def UpdateStock(ctx, product_id: str, new_stock_count: int) -> StockUpdateResult:

• Positional arguments to @rpc are the input parameter types in order — String, Integer

• _returns is the return type

• ctx is the spyne context (request metadata, headers, etc.) — always first, not an input parameter

• Parameter names in the function signature become the XSD element names in the WSDL

Step 3: Wire the Application

# soap_inventory/app.py
from spyne import Application
from spyne.protocol.soap import Soap11
from spyne.server.wsgi import WsgiApplication

from .service import InventoryService

# Build the spyne Application.
# - services: list of ServiceBase subclasses
# - tns: target namespace — becomes the WSDL targetNamespace
# - in_protocol / out_protocol: SOAP 1.1 or 1.2
application = Application(
    services=[InventoryService],
    tns="http://example.com/inventory",
    name="InventoryService",
    in_protocol=Soap11(validator="lxml"),   # Use lxml for XSD validation
    out_protocol=Soap11(),
)

wsgi_app = WsgiApplication(application)

Setting validator="lxml" on in_protocol enables XSD validation of incoming requests. Requests that do not match the schema return a SOAP Fault automatically without reaching your service code.

Step 4: Serve Over HTTP with Flask

# soap_inventory/server.py
from wsgiref.simple_server import make_server

from flask import Flask, request, Response

from .app import wsgi_app

flask_app = Flask(__name__)


@flask_app.route("/inventory", methods=["GET", "POST"])
def inventory_endpoint():
    # Flask does not natively speak WSGI the way spyne expects,
    # so we delegate to the spyne WSGI app for all requests.
    environ = request.environ
    response_started = []

    def start_response(status, headers, exc_info=None):
        response_started.append((status, headers))

    result = wsgi_app(environ, start_response)

    if response_started:
        status, headers = response_started[0]
        code = int(status.split(" ", 1)[0])
        resp = Response(
            response=b"".join(result),
            status=code,
            headers=dict(headers),
        )
        return resp

    return Response(b"".join(result), status=500)


if __name__ == "__main__":
    flask_app.run(host="0.0.0.0", port=8000, debug=True)

Alternative: Serve with wsgiref Directly

For local development and testing, Python's built-in wsgiref is simpler:

# serve_direct.py
from wsgiref.simple_server import make_server
from soap_inventory.app import wsgi_app

if __name__ == "__main__":
    server = make_server("0.0.0.0", 8000, wsgi_app)
    print("Serving SOAP on http://localhost:8000/")
    print("WSDL at: http://localhost:8000/?wsdl")
    server.serve_forever()

Start the server:

python serve_direct.py
# Serving SOAP on http://localhost:8000/
# WSDL at: http://localhost:8000/?wsdl

Viewing the Generated WSDL

Once the server is running, spyne automatically exposes the WSDL at ?wsdl:

curl http://localhost:8000/?wsdl

You can also load it in a browser or import it into SoapUI for visual inspection.

The WSDL generated by spyne maps directly to what you defined:

• Each @rpc method becomes an operation in the portType

• Each ComplexModel class becomes an xsd:complexType

• The tns argument to Application becomes the targetNamespace

Calling the Service

With the server running, here is a quick smoke test using zeep:

# quick_test.py
from decimal import Decimal
from zeep import Client

client = Client("http://localhost:8000/?wsdl")

# GetProduct
product = client.service.GetProduct(product_id="SKU-001")
print(f"Got: {product.Name} @ {product.Price}")

# CreateProduct
from zeep.helpers import serialize_object
new_product = client.service.CreateProduct(
    product_input={
        "ProductId": "SKU-003",
        "Name": "Super Widget",
        "Price": Decimal("49.99"),
        "InStock": True,
    }
)
print(f"Created: {new_product.ProductId}")

# UpdateStock
result = client.service.UpdateStock(
    product_id="SKU-001",
    new_stock_count=200,
)
print(f"Stock updated: {result.OldStockCount} -> {result.NewStockCount}")

Logging Incoming SOAP Messages

During development, I always add logging to inspect raw requests. In spyne, you handle this with an event handler on the application.

# soap_inventory/app.py (with logging)
import logging
from lxml import etree
from spyne import Application, event
from spyne.protocol.soap import Soap11
from spyne.server.wsgi import WsgiApplication

from .service import InventoryService

logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger("soap.inventory")


def _on_method_call(ctx):
    logger.debug(
        "Incoming SOAP call: service=%s operation=%s",
        ctx.service_class.__name__,
        ctx.function.__name__,
    )


def _on_method_return_object(ctx):
    logger.debug(
        "SOAP response: %s returned successfully",
        ctx.function.__name__,
    )


application = Application(
    services=[InventoryService],
    tns="http://example.com/inventory",
    name="InventoryService",
    in_protocol=Soap11(validator="lxml"),
    out_protocol=Soap11(),
)

application.event_manager.add_listener("method_call", _on_method_call)
application.event_manager.add_listener("method_return_object", _on_method_return_object)

wsgi_app = WsgiApplication(application)

Common spyne Pitfalls

Things that caught me when I first built a spyne service:

1. Parameter names in @rpc become XSD element names. If your method has product_id: str, the WSDL will have <product_id> in the input element. This matters when you need the WSDL to match an existing contract — use the exact capitalization expected.

2. ctx is not a typed parameter. spyne automatically passes the context as the first argument. Do not include it in @rpc(...) type list.

3. Array(SomeComplexModel) generates the correct unbounded list type. Using a plain Python list will not be serialised correctly — always use Array(T).

4. The _returns must be set explicitly. If missing, spyne generates a void return, and your response data will not appear in the WSDL or be serialised.

5. Use Decimal not float for monetary values. Float loses precision in XML serialisation.

What's Next

You have built a working SOAP service with spyne:

• Data models with ComplexModel

• Operations with @rpc

• XSD validation on incoming requests

• HTTP hosting with wsgiref and Flask

• Logging with application event handlers

In Part 4, we switch to the client side and use zeep to consume SOAP services robustly — handling complex types, managing sessions, caching WSDLs, and debugging transport issues.