Microsoft Semantic Kernel for Oracle EBS data access (Python)

Deployment Tips

• Infrastructure: Deploy this MCP server on Railway. Read our Step-by-Step Railway Deployment Guide.
• Security: This legacy system requires a VPN. Use NordLayer to bridge your cloud container to the internal network.

Microsoft Semantic Kernel for Oracle EBS Data Access (Python)

This guide details how to integrate Microsoft Semantic Kernel with Oracle E-Business Suite (EBS) using the Model Context Protocol (MCP).

By wrapping Oracle EBS queries in a FastMCP server, we create a secure, contract-based bridge. The Semantic Kernel agent consumes these tools natively, allowing it to perform complex financial or inventory lookups without needing direct, hard-coded SQL access in the agent logic.

🏗️ Architecture

1. Server (server.py): A Python-based FastMCP server running oracledb. It exposes typed tools (e.g., get_supplier_balance) via SSE.
2. Container: A Docker container enabling easy deployment to Railway, ECS, or Kubernetes.
3. Agent (agent.py): A Microsoft Semantic Kernel script that connects to the MCP server, registers the tools, and executes natural language requests.

---

🚀 Server Implementation (server.py)

The server uses python-oracledb in “Thin” mode, which eliminates the need for the heavy Oracle Instant Client in most cases.

import os
import oracledb
from mcp.server.fastmcp import FastMCP

# Initialize the MCP Server
mcp = FastMCP("OracleEBS-Gateway")

# Oracle Connection Config
# Ensure your container has network access (e.g. via NordLayer)
DB_USER = os.getenv("EBS_USER", "APPS")
DB_PASSWORD = os.getenv("EBS_PASSWORD", "password")
DB_DSN = os.getenv("EBS_DSN", "ebs.example.com:1521/VIS")

def get_connection():
    """Establishes a connection to the Oracle EBS database."""
    try:
        # Thin mode (no Instant Client required)
        return oracledb.connect(
            user=DB_USER,
            password=DB_PASSWORD,
            dsn=DB_DSN
        )
    except oracledb.Error as e:
        raise RuntimeError(f"Oracle Connection Failed: {str(e)}")

@mcp.tool()
def get_supplier_balance(supplier_name: str) -> str:
    """
    Retrieves the outstanding balance for a supplier in Oracle AP.
    Args:
        supplier_name: The name of the vendor (e.g., 'Acme Corp').
    """
    conn = get_connection()
    cursor = conn.cursor()
    try:
        # Standard Oracle AP query joining Invoices and Vendors
        query = """
            SELECT v.vendor_name, SUM(i.invoice_amount - NVL(i.amount_paid, 0))
            FROM ap_invoices_all i
            JOIN po_vendors v ON i.vendor_id = v.vendor_id
            WHERE UPPER(v.vendor_name) LIKE UPPER(:name)
            AND i.payment_status_flag != 'Y'
            GROUP BY v.vendor_name
        """
        cursor.execute(query, name=f"%{supplier_name}%")
        results = cursor.fetchall()

        if not results:
            return f"No outstanding balance found for '{supplier_name}'."

        response = [f"Balance Report for '{supplier_name}':"]
        for row in results:
            response.append(f"- {row[0]}: ${row[1]:,.2f}")

        return "\n".join(response)
    except Exception as e:
        return f"EBS Query Error: {str(e)}"
    finally:
        cursor.close()
        conn.close()

@mcp.tool()
def check_inventory(item_sku: str, organization_code: str = "M1") -> str:
    """
    Checks on-hand quantity for an item in a specific Inventory Org.
    """
    conn = get_connection()
    cursor = conn.cursor()
    try:
        query = """
            SELECT msi.segment1, sum(moq.transaction_quantity)
            FROM mtl_onhand_quantities moq
            JOIN mtl_system_items_b msi ON moq.inventory_item_id = msi.inventory_item_id
            JOIN mtl_parameters mp ON moq.organization_id = mp.organization_id
            WHERE msi.segment1 = :sku
            AND mp.organization_code = :org
            GROUP BY msi.segment1
        """
        cursor.execute(query, sku=item_sku, org=organization_code)
        result = cursor.fetchone()

        if result:
            return f"Item {result[0]}: {result[1]} units on hand in {organization_code}."
        return f"Item {item_sku} not found in org {organization_code}."
    except Exception as e:
        return f"EBS Query Error: {str(e)}"
    finally:
        cursor.close()
        conn.close()

if __name__ == "__main__":
    # MANDATORY: Bind to 0.0.0.0 for Docker compatibility
    mcp.run(transport='sse', host='0.0.0.0', port=8000)

---

🐳 Dockerfile

This Dockerfile is optimized for Railway and other container platforms. It exposes port 8000 and installs the necessary drivers.

# Use a slim Python base
FROM python:3.11-slim

# Install system utilities if needed
RUN apt-get update && apt-get install -y --no-install-recommends \
    curl \
    && rm -rf /var/lib/apt/lists/*

WORKDIR /app

# Install Python dependencies
RUN pip install "mcp[cli]" oracledb uvicorn

# Copy server code
COPY server.py .

# Environment variables (Override these in deployment)
ENV EBS_USER=APPS
ENV EBS_PASSWORD=changeme
ENV EBS_DSN=ebs.prod.local:1521/VIS

# Expose port 8000 for Railway
EXPOSE 8000

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

---

🔌 Agent Client (agent.py)

This client demonstrates how to connect Microsoft Semantic Kernel to the MCP server. It utilizes the ClientSession to dynamically discover tools and register them as kernel functions.

import asyncio
import os
from semantic_kernel import Kernel
from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion, OpenAIChatPromptExecutionSettings
from semantic_kernel.contents.chat_history import ChatHistory
from mcp import ClientSession, StdioServerParameters
from mcp.client.sse import sse_client

async def main():
    # 1. Configuration
    # Define MCP servers here. This pattern mimics the 'mcps' config found in other frameworks.
    mcps = ["http://localhost:8000/sse"]

    # Initialize Kernel
    kernel = Kernel()

    # Add Azure OpenAI or OpenAI Service
    service_id = "default"
    kernel.add_service(
        OpenAIChatCompletion(
            service_id=service_id,
            ai_model_id="gpt-4o",
            api_key=os.getenv("OPENAI_API_KEY", "sk-proj-...")
        )
    )

    print(f"🔌 Connecting to MCP Servers: {mcps}")

    # 2. Connect to the MCP Server
    # We use the first server in our list for this example
    async with sse_client(mcps[0]) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # 3. Dynamic Tool Discovery
            tools = await session.list_tools()
            print(f"✅ Found {len(tools.tools)} Oracle EBS tools.")

            # 4. Register Tools with Semantic Kernel
            # We wrap the generic MCP call in a closure for the Kernel to invoke
            for tool in tools.tools:
                async def tool_wrapper(params: dict = None, tool_name=tool.name):
                    # Default to empty dict if None
                    args = params if params else {}
                    result = await session.call_tool(tool_name, arguments=args)
                    return result.content[0].text

                # Register the function
                kernel.add_function(
                    plugin_name="OracleEBS",
                    function_name=tool.name,
                    function=tool_wrapper,
                    description=tool.description
                )
                print(f"   - Linked: {tool.name}")

            # 5. Execute Agent Logic
            chat_history = ChatHistory()
            user_query = "Check the supplier balance for 'Acme Corp' and also check stock for item 'AS54888'."
            chat_history.add_user_message(user_query)

            # Enable auto-invocation of the tools we just registered
            settings = OpenAIChatPromptExecutionSettings(
                service_id=service_id,
                tool_call_behavior="auto_invoke_kernel_functions"
            )

            print(f"\n🤖 User: {user_query}")
            print("⏳ Agent is thinking (querying Oracle)...")

            response = await kernel.invoke_prompt(
                prompt=user_query,
                settings=settings
            )

            print(f"\n💡 Response:\n{response}")

if __name__ == "__main__":
    asyncio.run(main())

Deployment Notes

1. Network Access: Ensure your container has network access (e.g. via NordLayer) if your Oracle EBS is behind a corporate firewall.
2. Concurrency: The agent.py script holds an open SSE connection while the Kernel thinks. This is efficient and allows for multi-turn conversations without reconnecting.
3. Security: The DB_USER should be a read-only Oracle user with limited scope (e.g., SELECT on AP_INVOICES_ALL, PO_VENDORS, etc.) to prevent accidental data modification by the LLM.

---

🛡️ Quality Assurance

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

Transparency: This page may contain affiliate links.