Skip to content
AgentRetrofit

AutoGen agents consuming and publishing to SOAP APIs (Python)

AutoGen Agents Consuming and Publishing to SOAP APIs (Python)

Section titled “AutoGen Agents Consuming and Publishing to SOAP APIs (Python)”

In the landscape of “Big Iron” enterprise software, SOAP (Simple Object Access Protocol) remains the dominant language of legacy communication. While modern AI agents (like AutoGen) speak JSON and REST natively, they often struggle when faced with the verbose XML payloads and WSDL contracts of 2000s-era ERPs.

This guide provides a FastMCP bridge that allows AutoGen agents to seamlessly consume and publish to legacy SOAP APIs using the industry-standard zeep library in Python. This enables your multi-agent workforce to perform tasks like checking inventory levels, updating shipping status, or creating invoices in systems like SAP ECC, Oracle EBS, or older Salesforce instances.

The Architecture: FastMCP + Zeep

Section titled “The Architecture: FastMCP + Zeep”

We use FastMCP to create a lightweight Model Context Protocol server. This server wraps the complex zeep logic, exposing simple, type-safe tools that AutoGen can call directly.

Features

Section titled “Features”
  1. WSDL Abstraction: The agent doesn’t need to parse XML; it just calls a function like get_order_status(order_id="123").
  2. Proxy Support: Built-in hooks for rotating proxies (essential for bypassing legacy firewall rate limits or accessing internal networks).
  3. Error Translation: Converts cryptic SOAP Faults into clear, actionable error messages for the LLM.

1. The Code (server.py)

Section titled “1. The Code (server.py)”

This bridge exposes a tool to interact with a hypothetical “LegacyOrderService”. In a real deployment, you would replace the WSDL URL with your internal enterprise endpoint.

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


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


# CONSTANTS
# In production, this would be an environment variable or internal network path
WSDL_URL = "http://webservices.oorsprong.org/websamples.countryinfo/CountryInfoService.wsdl"
# NOTE: The public WSDL above is for testing. Replace with your internal ERP WSDL.


@mcp.tool()
def soap_service_query(method: str, params: str) -> str:
    """
    Executes a SOAP operation against the Legacy Order Service.


    Args:
        method: The name of the SOAP operation to call (e.g., 'FullCountryInfo', 'InventoryCheck').
        params: A JSON string of parameters required by the operation.
                Example: '{"sCountryISOCode": "US"}'


    Returns:
        A stringified JSON response from the SOAP service or an error message.
    """


    # For production, inject BrightData proxy URL here
    # proxies = {
    #    'http': 'http://user:pass@brd.superproxy.io:22225',
    #    'https': 'http://user:pass@brd.superproxy.io:22225'
    # }


    # Setup session with optional proxies
    session = requests.Session()
    # session.proxies = proxies # Uncomment to enable rotation


    # Configure Zeep Transport
    transport = Transport(session=session)


    # 'strict=False' helps handle non-compliant legacy XML often found in old ERPs
    settings = Settings(strict=False, xml_huge_tree=True)


    try:
        # Initialize Client
        client = Client(wsdl=WSDL_URL, transport=transport, settings=settings)


        # Check if method exists
        if not hasattr(client.service, method):
            return f"Error: Method '{method}' not found in WSDL."


        # Parse JSON params to dict
        try:
            param_dict = json.loads(params)
        except json.JSONDecodeError:
            return "Error: 'params' argument must be a valid JSON string."


        # Dynamically retrieve and call the method
        service_method = getattr(client.service, method)


        # Execute the SOAP call
        # Unpacking **param_dict passes keys as named arguments to the SOAP function
        result = service_method(**param_dict)


        # Serialize result to JSON for the agent
        # Zeep results are often complex objects, usually specific helpers are needed
        # to serialize fully, but str() or zeep.helpers.serialize_object works for simple cases.
        from zeep.helpers import serialize_object
        return json.dumps(serialize_object(result), default=str)


    except Fault as fault:
        return f"SOAP Fault: {fault.message} (Code: {fault.code})"
    except Exception as e:
        return f"System Error: {str(e)}"


if __name__ == "__main__":
    mcp.run()

Key Implementation Details

Section titled “Key Implementation Details”
  • Dynamic Dispatch: The soap_service_query tool is generic. It takes a method name and a JSON string of parameters. This allows the AutoGen agent to explore the API (if you were to add a list_methods tool) or call any endpoint without you writing a wrapper for every single function.
  • JSON Handling: AutoGen agents speak JSON. We force the input params to be a JSON string and convert the XML output back to JSON using zeep.helpers.serialize_object.
  • Proxy Injection: The commented-out proxy section is critical for enterprise environments where the legacy SOAP server is behind a firewall that requires a specific egress IP.

2. The Container (Dockerfile)

Section titled “2. The Container (Dockerfile)”

To deploy this as a microservice compatible with Railway, Kubernetes, or AutoGen Studio, use the following Dockerfile.

# Use a slim Python base for speed and security
FROM python:3.11-slim


# Set working directory
WORKDIR /app


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


# Copy requirements (inline for simplicity in this guide, or use a requirements.txt)
# We install fastmcp, zeep, and requests
RUN pip install --no-cache-dir fastmcp zeep requests


# Copy the server code
COPY server.py .


# Expose port 8000 for Railway/MCP compatibility
EXPOSE 8000


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

3. Integration with AutoGen

Section titled “3. Integration with AutoGen”

Once your Docker container is running (e.g., at http://localhost:8000 or a deployed URL), you can connect an AutoGen agent to it.

If you are using AutoGen Studio or the standard AutoGen library, you typically define the tool interface manually or use an MCP client bridge.

Example AutoGen Configuration (Python):

from autogen import UserProxyAgent, AssistantAgent


# 1. Define the tool signature (mirrors the MCP tool)
def soap_query_tool(method: str, params: str) -> str:
    # In a real setup, this function would call your MCP server API
    # via HTTP POST /mcp/execute or similar, depending on your MCP client.
    # For local dev, you might call the function directly if running in same process.
    pass


# 2. Configure the Assistant
config_list = [{"model": "gpt-4-turbo", "api_key": "YOUR_KEY"}]


assistant = AssistantAgent(
    name="LegacySystemSpecialist",
    system_message="""You are an expert in legacy ERP systems.
    Use the 'soap_service_query' tool to interact with the SOAP API.
    Always pass parameters as a valid JSON string.""",
    llm_config={
        "config_list": config_list,
        "functions": [
            {
                "name": "soap_service_query",
                "description": "Call a SOAP API method.",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "method": {"type": "string", "description": "SOAP operation name"},
                        "params": {"type": "string", "description": "JSON arguments"}
                    },
                    "required": ["method", "params"]
                }
            }
        ]
    }
)


user_proxy = UserProxyAgent(
    name="User_Proxy",
    code_execution_config={"work_dir": "coding"},
    human_input_mode="NEVER"
)


# 3. Register the tool execution (Mocking the MCP connection for this snippet)
# In production, connect this to the running FastMCP server.
user_proxy.register_function(
    function_map={"soap_service_query": soap_query_tool}
)


# 4. Initiate Chat
user_proxy.initiate_chat(
    assistant,
    message="Check the full country info for ISO code 'US' using the SOAP service."
)

Troubleshooting Common Errors

Section titled “Troubleshooting Common Errors”
  • lxml.etree.XMLSyntaxError: The legacy server is returning malformed XML. Ensure Settings(strict=False) is set in server.py.
  • ConnectionRefusedError: The SOAP server is likely behind a firewall. Enable the proxies configuration in server.py and route traffic through a static IP proxy.
  • TypeError: Object of type X is not JSON serializable: Zeep returns complex Python objects. Ensure you use zeep.helpers.serialize_object before dumping to JSON.

🛡️ Quality Assurance

Section titled “🛡️ Quality Assurance”
  • Status: ✅ Verified
  • Environment: Python 3.11
  • Auditor: AgentRetrofit CI/CD

Transparency: This page may contain affiliate links.