Automating Oracle EBS with CrewAI agents using `cx_Oracle`

Automating Oracle EBS with CrewAI and Python (cx_Oracle)

This guide details how to bridge modern CrewAI agents with legacy Oracle E-Business Suite (EBS) infrastructure.

While modern SaaS APIs are JSON-native, Oracle EBS often relies on TNS protocols and heavy SQL interactions. We will create a Model Context Protocol (MCP) server that acts as a translator, allowing your AI agents to query EBS financials, inventory, and order status directly.

🏗️ Architecture

We use FastMCP to create a lightweight bridge. The agent does not need to know PL/SQL; it simply asks tools like get_inventory_levels or check_invoice_status to perform the work.

The Stack

Protocol: MCP (Model Context Protocol) over SSE (Server-Sent Events).
Driver: cx_Oracle (Standard Python interface for Oracle DB).
Container: Docker (Debian-based Python slim image).

🚀 The Integration Code

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

This server exposes Oracle EBS data as agent-ready tools. It handles the TNS connection and formats the output into strings the LLM can understand.

from fastmcp import FastMCP
import cx_Oracle
import os
import json

# Ensure your container has network access (e.g. via NordLayer)
# This is critical for reaching on-prem Oracle instances from the cloud.

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

# Helper to get database connection
def get_connection():
    # Construct DSN from environment variables
    # Oracle EBS usually requires a Service Name (SID) or Service Name
    dsn = cx_Oracle.makedsn(
        os.getenv("ORACLE_HOST", "localhost"),
        os.getenv("ORACLE_PORT", "1521"),
        service_name=os.getenv("ORACLE_SERVICE", "VIS")
    )

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

@mcp.tool()
def get_ebs_inventory(item_number: str, organization_code: str) -> str:
    """
    Retrieves on-hand quantity for an item in a specific Oracle Inventory Organization.
    Useful for checking stock before placing orders.
    """
    sql = """
        SELECT moq.transaction_quantity, msi.description
        FROM mtl_onhand_quantities moq
        JOIN mtl_system_items_b msi
          ON moq.inventory_item_id = msi.inventory_item_id
          AND moq.organization_id = msi.organization_id
        JOIN mtl_parameters mp
          ON moq.organization_id = mp.organization_id
        WHERE msi.segment1 = :item_num
          AND mp.organization_code = :org_code
    """

    try:
        conn = get_connection()
        cursor = conn.cursor()
        cursor.execute(sql, item_num=item_number, org_code=organization_code)
        rows = cursor.fetchall()

        if not rows:
            return f"No inventory found for Item {item_number} in Org {organization_code}."

        # Format for the Agent
        total_qty = sum(row[0] for row in rows)
        description = rows[0][1]

        cursor.close()
        conn.close()

        return json.dumps({
            "item": item_number,
            "description": description,
            "organization": organization_code,
            "total_on_hand": total_qty
        })

    except cx_Oracle.DatabaseError as e:
        error, = e.args
        return f"Oracle EBS Error: {error.message}"

@mcp.tool()
def get_invoice_status(invoice_num: str) -> str:
    """
    Checks the payment status of an AP Invoice in Oracle EBS.
    """
    sql = """
        SELECT invoice_num, invoice_amount, amount_paid, payment_status_flag
        FROM ap_invoices_all
        WHERE invoice_num = :inv_num
    """
    try:
        conn = get_connection()
        cursor = conn.cursor()
        cursor.execute(sql, inv_num=invoice_num)
        row = cursor.fetchone()

        cursor.close()
        conn.close()

        if row:
            return f"Invoice {row[0]}: Total {row[1]}, Paid {row[2]}, Status: {row[3]}"
        return "Invoice not found."
    except Exception as e:
        return f"Error querying AP: {str(e)}"

if __name__ == "__main__":
    # Must bind to 0.0.0.0 for Docker visibility
    mcp.run(transport='sse', host='0.0.0.0', port=8000)

2. The Dockerfile

Using cx_Oracle requires the Oracle Instant Client libraries. This Dockerfile handles the complex dependency chain so you don’t have to manually install libaio on your machine.

# Use a lightweight Python base
FROM python:3.10-slim

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

# Setup Oracle Instant Client (Basic Lite)
# Adjust version numbers as needed for your specific Oracle DB version
WORKDIR /opt/oracle
RUN wget https://download.oracle.com/otn_software/linux/instantclient/2112000/instantclient-basiclite-linux.x64-21.12.0.0.0dbru.zip \
    && unzip instantclient-basiclite-linux.x64-21.12.0.0.0dbru.zip \
    && rm instantclient-basiclite-linux.x64-21.12.0.0.0dbru.zip \
    && sh -c "echo /opt/oracle/instantclient_21_12 > /etc/ld.so.conf.d/oracle-instantclient.conf" \
    && ldconfig

# Set up the application
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy source code
COPY server.py .

# Expose the SSE port
EXPOSE 8000

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

requirements.txt:

fastmcp
cx_Oracle
uvicorn

🔌 Connecting CrewAI

Once your Docker container is running, the CrewAI agent connects to the SSE endpoint. The agent will automatically discover the get_ebs_inventory and get_invoice_status tools.

Docker Run Command

docker build -t oracle-ebs-mcp .
docker run -d \
  -p 8000:8000 \
  -e ORACLE_HOST="192.168.1.50" \
  -e ORACLE_PORT="1521" \
  -e ORACLE_SERVICE="PROD" \
  -e ORACLE_USER="APPS" \
  -e ORACLE_PASSWORD="YOUR_PASSWORD" \
  oracle-ebs-mcp

CrewAI Agent Configuration

from crewai import Agent, Task, Crew
from langchain_openai import ChatOpenAI

# 1. Connect to the MCP Server running in Docker
# The URL must match the exposed port in Docker
ebs_mcp_source = ["http://localhost:8000/sse"]

# 2. Define the Agent
oracle_admin = Agent(
    role='Oracle EBS Specialist',
    goal='Retrieve accurate financial and inventory data from the legacy ERP',
    backstory="You are a veteran ERP administrator. You know how to interpret Oracle error codes.",
    llm=ChatOpenAI(model="gpt-4o", temperature=0),
    # Connect the MCP source here
    mcps=ebs_mcp_source
)

# 3. Define the Task
audit_task = Task(
    description="Check inventory for item 'AS54888' in Org 'M1'. If stock is below 10, check invoice status for 'INV-99887'.",
    expected_output="A summary of inventory levels and invoice payment status.",
    agent=oracle_admin
)

# 4. Run the Crew
crew = Crew(
    agents=[oracle_admin],
    tasks=[audit_task],
    verbose=True
)

result = crew.kickoff()
print(result)

⚠️ Common Troubleshooting

DPI-1047 Error: This means Python cannot find the Oracle Instant Client libraries. Ensure ldconfig was run in the Dockerfile and libaio1 is installed.
TNS:protocol adapter error: The container cannot reach the Oracle Host. If Oracle is on your host machine, use host.docker.internal instead of localhost. If it is on a VPN, ensure the container shares the VPN network context.
Authentication: Oracle EBS passwords often expire. Ensure the user APPS (or your custom read-only user) has an active, unlocked account.

🛡️ Quality Assurance

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

Transparency: This page may contain affiliate links.