Skip to content

Integrating Semantic Kernel with Legacy Salesforce SOAP WSDLs

Integrating Semantic Kernel with Legacy Salesforce SOAP WSDLs

Section titled “Integrating Semantic Kernel with Legacy Salesforce SOAP WSDLs”

While Microsoft Semantic Kernel (C#) is a powerful orchestrator, connecting it to legacy SOAP protocols (like the 15-year-old Salesforce enterprise.wsdl) is often best handled by a dedicated “Bridge” service.

This guide implements a FastMCP server in Python to handle the heavy lifting of WSDL parsing and XML serialization using zeep. It exposes these capabilities as Model Context Protocol (MCP) tools. While your main application logic might reside in .NET, we use a standard Python Agent (CrewAI) here to validate the integration and demonstrate the mcps connectivity pattern.

This server acts as a proxy. It loads the massive Salesforce WSDL once and translates clean tool calls into complex SOAP envelopes.

Requirements:

  • pip install fastmcp zeep
  • A valid enterprise.wsdl file in the same directory.
import os
from fastmcp import FastMCP
from zeep import Client
from zeep.transports import Transport
from requests import Session
# Initialize FastMCP
mcp = FastMCP("SalesforceSoapBridge")
# Global client storage
sf_client = None
session_header = None
# For production, inject BrightData proxy URL here
# proxies = {
# 'http': 'http://user:[email protected]:22225',
# 'https': 'http://user:[email protected]:22225'
# }
proxies = None
def get_soap_client(wsdl_path: str = "enterprise.wsdl"):
"""Singleton helper to load the heavy WSDL once."""
global sf_client
if not sf_client:
session = Session()
if proxies:
session.proxies.update(proxies)
transport = Transport(session=session)
# Ensure enterprise.wsdl is available in the Docker container
sf_client = Client(wsdl=wsdl_path, transport=transport)
return sf_client
@mcp.tool()
def salesforce_login(username: str, password: str, security_token: str) -> str:
"""
Logs into Salesforce SOAP API.
Returns success message and sets internal session state.
"""
global session_header, sf_client
try:
client = get_soap_client()
# Salesforce SOAP often requires password + token concatenation
full_password = f"{password}{security_token}"
# Execute login call
result = client.service.login(username=username, password=full_password)
# Store session header for subsequent calls
session_id = result['sessionId']
server_url = result['serverUrl']
session_header = {
'SessionHeader': {
'sessionId': session_id
}
}
# Rebind the service to the URL returned by login
client.service._binding_options['address'] = server_url
return f"Login successful. Connected to {server_url}"
except Exception as e:
return f"Login Failed: {str(e)}"
@mcp.tool()
def execute_soql(query_string: str) -> str:
"""
Executes a SOQL query against the Salesforce SOAP API using the active session.
Example: 'SELECT Id, Name FROM Account LIMIT 5'
"""
global sf_client, session_header
if not session_header:
return "Error: Not logged in. Call salesforce_login first."
try:
# Zeep handles the SOAP envelope construction
result = sf_client.service.query(
queryString=query_string,
_soapheaders=session_header
)
records = []
if result and result.records:
for record in result.records:
# Basic dict extraction, ignoring XML metadata
data = {k: v for k, v in record.items() if k != 'type' and v is not None}
records.append(data)
return str(records)
except Exception as e:
return f"SOQL Error: {str(e)}"
if __name__ == "__main__":
mcp.run(transport='sse', host='0.0.0.0', port=8000)

We containerize the Python bridge to isolate the legacy XML dependencies.

Dockerfile

FROM python:3.11-slim
WORKDIR /app
# Install system dependencies for lxml (required by zeep)
RUN apt-get update && apt-get install -y \
libxml2-dev \
libxslt-dev \
gcc \
&& rm -rf /var/lib/apt/lists/*
# Install libraries
RUN pip install --no-cache-dir fastmcp zeep requests
# Copy application code
COPY server.py .
# Ensure you have the WSDL file in the build context
COPY enterprise.wsdl .
# EXPOSE 8000 for Railway/Docker networking
EXPOSE 8000
CMD ["python", "server.py"]

To verify the MCP server is functioning correctly before integrating it into a larger Semantic Kernel C# application, we use a Python CrewAI agent. This serves as the acceptance test for the “Bridge.”

Prerequisites:

  • pip install crewai
  • The Docker container running on port 8000.
from crewai import Agent, Task, Crew
# 1. Define the Agent with MCP Connectivity
# The 'mcps' argument connects the agent to our Dockerized SOAP bridge
salesforce_agent = Agent(
role="Salesforce Legacy Specialist",
goal="Retrieve account data from the legacy Salesforce SOAP interface",
backstory="You are an expert in legacy ERP systems. You use a specialized bridge to talk to SOAP endpoints.",
# Connects to the server.py running in Docker
mcps=["http://localhost:8000/sse"],
verbose=True
)
# 2. Define the Task
# The agent will automatically figure out it needs to login before querying
sync_task = Task(
description=(
"1. Login to Salesforce using username '[email protected]', password 'Password123', and token 'XYZTOKEN'. "
"2. Query the first 5 Accounts (Id, Name). "
"3. Return the names found."
),
expected_output="A list of Account names from Salesforce.",
agent=salesforce_agent
)
# 3. Execute
my_crew = Crew(
agents=[salesforce_agent],
tasks=[sync_task]
)
if __name__ == "__main__":
result = my_crew.kickoff()
print("### Task Result ###")
print(result)
  1. Start the Server:
    Terminal window
    docker build -t sf-bridge .
    docker run -p 8000:8000 sf-bridge
  2. Run the Agent:
    Terminal window
    python agent.py

The agent will discover the salesforce_login and execute_soql tools via the MCP protocol, execute the login, manage the session state on the server side, and then retrieve the data—abstracting away the XML complexity entirely.


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

Transparency: This page may contain affiliate links.