Microsoft Semantic Kernel for Oracle EBS data access (Python)

Microsoft Semantic Kernel for Oracle EBS Data Access (Python)

As enterprises race to adopt AI agents, a massive friction point has emerged: connecting modern orchestration frameworks like Microsoft Semantic Kernel to legacy ERPs like Oracle E-Business Suite (EBS).

Oracle EBS typically runs the financial and supply chain backbone of large organizations. It relies on complex schemas, proprietary protocols (SQL*Net), and strict access controls. Directly connecting an LLM to this is dangerous and technically difficult.

This guide provides a “Retrofit” solution: a lightweight Model Context Protocol (MCP) server that acts as a secure gateway. It allows Semantic Kernel agents (or any MCP-compliant agent) to query Oracle EBS data safely using standard Python libraries, without requiring a massive middleware overhaul.

🛠️ The Architecture

We will build an MCP server that exposes specific, controlled tools to your AI Agent.

The Agent: Built with Microsoft Semantic Kernel (Python or C#).
The Protocol: MCP (Model Context Protocol).
The Bridge: A fastmcp server running inside a Docker container.
The Legacy System: Oracle EBS (Database Layer).

Prerequisites

Python 3.10+
Oracle Database Access: You need a read-only service account for your EBS database.
Network Access: The container must be able to reach the Oracle host (port 1521 usually).

💻 The Bridge Code (`server.py`)

This code uses python-oracledb (the upgraded version of cx_Oracle) which runs in “Thin” mode by default, removing the need for heavy Oracle Instant Client libraries for most modern EBS versions (Database 12c+).

import os
import oracledb
from fastmcp import FastMCP
from contextlib import contextmanager

# Initialize FastMCP
mcp = FastMCP("OracleEBS")

# Configuration via Environment Variables
DB_USER = os.getenv("ORACLE_USER", "APPS_READONLY")
DB_PASSWORD = os.getenv("ORACLE_PASSWORD")
DB_DSN = os.getenv("ORACLE_DSN", "ebs.example.com:1521/VIS")

# Ensure your container has network access (e.g. via NordLayer, Tailscale, or VPC peering)
# if the Oracle EBS instance is behind a corporate firewall.

@contextmanager
def get_db_connection():
    """Context manager for Oracle DB connections."""
    connection = None
    try:
        # Defaults to Thin mode.
        # For very old EBS versions (pre-12c), you might need to enable Thick mode
        # and install Oracle Instant Client in the Dockerfile.
        connection = oracledb.connect(
            user=DB_USER,
            password=DB_PASSWORD,
            dsn=DB_DSN
        )
        yield connection
    except oracledb.Error as e:
        raise RuntimeError(f"Oracle Connection Error: {e}")
    finally:
        if connection:
            connection.close()

@mcp.tool()
def get_ebs_table_metadata(table_name: str) -> str:
    """
    Retrieves the column definitions for a specific Oracle EBS table.
    Useful for the agent to understand the schema before writing queries.

    Args:
        table_name: The name of the table (e.g., PO_HEADERS_ALL).
    """
    sql = """
        SELECT column_name, data_type, data_length
        FROM all_tab_columns
        WHERE table_name = :table_name
        AND owner = 'APPS'
    """

    try:
        with get_db_connection() as conn:
            with conn.cursor() as cursor:
                cursor.execute(sql, table_name=table_name.upper())
                rows = cursor.fetchall()

                if not rows:
                    return f"No metadata found for table {table_name}. Check spelling or permissions."

                # Format as markdown for better LLM comprehension
                md = f"### Schema for {table_name.upper()}\n| Column | Type | Length |\n|---|---|---|\n"
                for row in rows:
                    md += f"| {row[0]} | {row[1]} | {row[2]} |\n"
                return md
    except Exception as e:
        return f"Error retrieving metadata: {str(e)}"

@mcp.tool()
def query_ebs_data(sql_query: str) -> str:
    """
    Executes a READ-ONLY SQL query against the Oracle EBS database.
    RESTRICTION: Only SELECT statements are allowed.

    Args:
        sql_query: The SQL SELECT statement to execute.
    """
    # Basic SQL Injection / Safety sanity check
    normalized_query = sql_query.strip().upper()
    if not normalized_query.startswith("SELECT"):
        return "Security Alert: Only SELECT statements are allowed via this interface."

    if any(keyword in normalized_query for keyword in ["DROP", "DELETE", "UPDATE", "INSERT", "ALTER", "TRUNCATE"]):
        return "Security Alert: Destructive commands are strictly prohibited."

    try:
        with get_db_connection() as conn:
            with conn.cursor() as cursor:
                # Limit rows to prevent context window overflow
                if "ROWNUM" not in normalized_query and "FETCH FIRST" not in normalized_query:
                    sql_query = f"SELECT * FROM ({sql_query}) WHERE ROWNUM <= 50"

                cursor.execute(sql_query)
                columns = [col[0] for col in cursor.description]
                rows = cursor.fetchall()

                if not rows:
                    return "Query returned no results."

                # Convert to simple JSON-like string format
                result_list = []
                for row in rows:
                    # Create a dictionary for the row
                    row_dict = dict(zip(columns, row))
                    # Handle non-serializable types (like LOBs or Dates) simply
                    clean_dict = {k: str(v) for k, v in row_dict.items()}
                    result_list.append(clean_dict)

                return str(result_list)

    except oracledb.Error as e:
        error_obj = e.args[0]
        return f"Oracle Database Error {error_obj.code}: {error_obj.message}"
    except Exception as e:
        return f"System Error: {str(e)}"

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

🐳 Containerization (`Dockerfile`)

We use a slim Python image to keep deployment fast.

# Use an official lightweight Python image
FROM python:3.11-slim

# Set working directory
WORKDIR /app

# Install system dependencies if needed (e.g., libaio1 is often needed for Oracle libs)
RUN apt-get update && apt-get install -y --no-install-recommends \
    libaio1 \
    && rm -rf /var/lib/apt/lists/*

# Install Python dependencies
# fastmcp: The MCP server framework
# oracledb: The official Oracle DB driver
RUN pip install --no-cache-dir fastmcp oracledb

# Copy the server code
COPY server.py .

# Expose the port for the MCP server
# CRITICAL: This port must be mapped when running the container
EXPOSE 8000

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

🚀 Deployment & Integration

1. Build and Run

docker build -t ebs-mcp-agent .
docker run -d -p 8000:8000 \
  -e ORACLE_USER="APPS_READONLY" \
  -e ORACLE_PASSWORD="your_secure_password" \
  -e ORACLE_DSN="ebs-prod.corp.local:1521/VIS" \
  --name ebs_agent \
  ebs-mcp-agent

2. Connecting Semantic Kernel

In your Semantic Kernel application code, you treat this MCP server as a plugin. The agent effectively gains the “skill” of querying Oracle EBS.

The agent will now self-correct common Oracle errors. For example, if it receives ORA-00942: table or view does not exist, it can use the get_ebs_table_metadata tool to check if it’s querying PO_HEADERS instead of the correct PO_HEADERS_ALL.

3. Troubleshooting “Big Iron” Errors

ORA-12541: TNS:no listener: The agent cannot reach the host. Check your VPN/Container network settings.
ORA-28000: the account is locked: EBS accounts lock frequently. Ensure you are using a dedicated service account, not a personal user login.
Slow Queries: Oracle EBS tables have millions of rows. The code limits results to 50 rows by default; advise the agent to write specific WHERE clauses (e.g., WHERE CREATION_DATE > SYSDATE - 1).

🛡️ Quality Assurance

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

Transparency: This page may contain affiliate links.