Automating Oracle EBS with CrewAI agents using `cx_Oracle`

Automating Oracle EBS with CrewAI Agents (Python & cx_Oracle)

In the “Retrofit Era,” connecting modern AI agents to Oracle E-Business Suite (EBS) is a common but daunting requirement. While modern SaaS platforms offer clean REST APIs, Oracle EBS (often version 12.1 or 12.2) relies heavily on direct database connections, PL/SQL packages, and strict TNS networking layers.

This guide provides a production-ready Model Context Protocol (MCP) server that allows CrewAI, LangGraph, or OpenAI Operator agents to safely interact with your Oracle EBS instance.

🛠️ The Challenge: The “Thick Client” Problem

Oracle EBS doesn’t just “talk” SQL; it often requires the Oracle Call Interface (OCI), which means you cannot simply pip install a driver and expect it to work in a serverless container. You need:

The OS-Level Drivers: Oracle Instant Client binaries (libclntsh.so).
The Network Layer: TNS resolution and firewall access (usually via VPN).
The Python Wrapper: cx_Oracle (the industry standard for legacy support) or oracledb.

We will solve this by containerizing the entire stack, exposing a clean MCP interface that your agents can query without knowing the underlying complexity.

⚡ The Solution: FastMCP Bridge

We will build a Dockerized MCP server that exposes two critical EBS functions to your agents:

check_inventory_levels: Checks stock in specific organizations (e.g., MTL_ONHAND_QUANTITIES).
get_order_status: Retrieves order headers from Order Management (OE_ORDER_HEADERS_ALL).

1. The MCP Server Code (`server.py`)

This Python script uses fastmcp to define the tools. It manages the Oracle connection pool and executes read-only queries safely.

import os
import cx_Oracle
from fastmcp import FastMCP

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

# Ensure your container has network access (e.g. via NordLayer)
# CRITICAL: Oracle EBS is rarely exposed to the public internet.
# This container must run inside your VPN or VPC.

def get_connection():
    """Establishes a connection to the Oracle EBS Database."""
    # In production, use AWS Secrets Manager or Vault
    dsn = cx_Oracle.makedsn(
        os.getenv("EBS_HOST", "ebs.example.com"),
        os.getenv("EBS_PORT", "1521"),
        service_name=os.getenv("EBS_SERVICE", "VIS")
    )

    return cx_Oracle.connect(
        user=os.getenv("EBS_USER"),
        password=os.getenv("EBS_PASSWORD"),
        dsn=dsn,
        encoding="UTF-8"
    )

@mcp.tool()
def check_inventory_levels(item_number: str, org_code: str) -> str:
    """
    Checks on-hand quantity for a specific item in a specific EBS Organization.

    Args:
        item_number: The EBS item part number (e.g., 'AS54888').
        org_code: The 3-letter organization code (e.g., 'M1', 'V1').
    """
    sql = """
        SELECT
            msi.segment1 as item,
            mp.organization_code,
            SUM(moq.transaction_quantity) as total_qty,
            msi.description
        FROM
            inv.mtl_onhand_quantities_detail moq
            JOIN inv.mtl_system_items_b msi ON moq.inventory_item_id = msi.inventory_item_id
                AND moq.organization_id = msi.organization_id
            JOIN inv.mtl_parameters mp ON moq.organization_id = mp.organization_id
        WHERE
            msi.segment1 = :item_number
            AND mp.organization_code = :org_code
        GROUP BY
            msi.segment1, mp.organization_code, msi.description
    """

    conn = None
    try:
        conn = get_connection()
        cursor = conn.cursor()
        cursor.execute(sql, item_number=item_number, org_code=org_code)
        result = cursor.fetchone()

        if result:
            return f"Item {result[0]} ({result[3]}) in Org {result[1]}: {result[2]} units on hand."
        else:
            return f"No inventory found for Item {item_number} in Org {org_code}."

    except cx_Oracle.Error as e:
        return f"Oracle EBS Error: {e}"
    finally:
        if conn:
            conn.close()

@mcp.tool()
def get_order_status(order_number: int) -> str:
    """
    Retrieves the status and total value of a Sales Order.

    Args:
        order_number: The numerical sales order ID.
    """
    sql = """
        SELECT
            header_id,
            order_number,
            flow_status_code,
            ordered_date,
            transactional_curr_code
        FROM
            ont.oe_order_headers_all
        WHERE
            order_number = :order_number
    """

    conn = None
    try:
        conn = get_connection()
        cursor = conn.cursor()
        cursor.execute(sql, order_number=order_number)
        result = cursor.fetchone()

        if result:
            return (f"Order #{result[1]} Status: {result[2]} | "
                    f"Date: {result[3]} | Currency: {result[4]}")
        else:
            return f"Order #{order_number} not found."

    except cx_Oracle.Error as e:
        return f"Oracle EBS Error: {e}"
    finally:
        if conn:
            conn.close()

if __name__ == "__main__":
    # Runs the MCP server on port 8000
    mcp.run()

2. The Dockerfile (`Dockerfile`)

This is the most critical part. We must install the Oracle Instant Client libraries so cx_Oracle can function.

# Use a slim Python image to keep size down, but Debian-based for easy apt usage
FROM python:3.11-slim

# Set environment variables for Oracle Instant Client
ENV ORACLE_HOME=/opt/oracle/instantclient_19_19
ENV LD_LIBRARY_PATH=$ORACLE_HOME:$LD_LIBRARY_PATH
ENV PATH=$ORACLE_HOME:$PATH

# Install system dependencies
# libaio1 is REQUIRED for Oracle Instant Client
RUN apt-get update && apt-get install -y \
    libaio1 \
    wget \
    unzip \
    && rm -rf /var/lib/apt/lists/*

# Download and Install Oracle Instant Client Basic Lite
# Note: In production, host this zip on your own internal S3/Artifactory to avoid rate limits
WORKDIR /opt/oracle
RUN wget https://download.oracle.com/otn_software/linux/instantclient/1919000/instantclient-basiclite-linux.x64-19.19.0.0.0dbru.zip \
    && unzip instantclient-basiclite-linux.x64-19.19.0.0.0dbru.zip \
    && rm -f instantclient-basiclite-linux.x64-19.19.0.0.0dbru.zip

# Setup Application
WORKDIR /app
COPY server.py .

# Install Python dependencies
# We pin fastmcp and cx_Oracle for stability
RUN pip install --no-cache-dir fastmcp cx_Oracle

# EXPOSE the port for Railway/Docker
EXPOSE 8000

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

🚀 How to Deploy & Use

1. Build and Run

Build the container locally to verify the Oracle libraries load correctly:

docker build -t ebs-mcp .
docker run -p 8000:8000 --env-file .env ebs-mcp

2. Connect CrewAI

Once your Docker container is running (locally or on a private cloud service like Railway/AWS ECS), you can connect your CrewAI agents using the SSE (Server-Sent Events) transport.

from crewai import Agent, Task, Crew
from crewai_tools import MCPServerTool

# Connect to the Dockerized MCP Server
ebs_tool = MCPServerTool(
    server_url="http://localhost:8000/sse", # Or your deployed URL
    tool_name="check_inventory_levels"
)

supply_chain_agent = Agent(
    role='Supply Chain Manager',
    goal='Ensure stock availability',
    backstory='You monitor the legacy Oracle ERP system to prevent stockouts.',
    tools=[ebs_tool],
    verbose=True
)

task = Task(
    description="Check if we have enough stock of item 'AS54888' in warehouse 'M1'.",
    agent=supply_chain_agent
)

crew = Crew(agents=[supply_chain_agent], tasks=[task])
result = crew.kickoff()
print(result)

⚠️ Troubleshooting “Big Iron” Errors

DPI-1047: Cannot locate a 64-bit Oracle Client library: This means libaio1 is missing or LD_LIBRARY_PATH is not set in the Dockerfile. Use the Dockerfile provided above exactly.
ORA-12170: TNS:Connect timeout: The agent cannot reach the EBS database. Ensure your container is running inside the corporate VPN or VPC. You may need a sidecar container (like NordLayer or Tailscale) if deploying on public cloud infrastructure.

🛡️ Quality Assurance

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

Transparency: This page may contain affiliate links.