Integrating LangGraph with Oracle EBS workflows using `cx_Oracle`

In the enterprise automation landscape, Oracle E-Business Suite (EBS) remains the backbone of supply chain and financial operations for thousands of companies. However, modern agentic frameworks like LangGraph struggle to interact with EBS due to its reliance on PL/SQL, proprietary TNS protocols, and complex schema structures.

This guide provides a production-ready Model Context Protocol (MCP) server that acts as a secure gateway between your LangGraph agents and Oracle EBS. Using the standard cx_Oracle library, this server allows agents to query workflow statuses, check inventory, and trigger stored procedures without direct database exposure.

🛠️ Architectural Overview

This integration follows the “Sidecar Pattern”:

The Agent (LangGraph): Reasons about the task (e.g., “Check status of PO #1023”).
The Bridge (FastMCP): An independent container running Python and the Oracle Instant Client.
The Legacy System (Oracle EBS): Executes the SQL/PLSQL and returns raw data.

Why `cx_Oracle`?

While newer drivers exist, cx_Oracle remains the certified standard for many legacy EBS installations (R12.1/R12.2) requiring specific TNS configuration and OCI (Oracle Call Interface) features not fully supported by thin clients.

💻 The Bridge Code

This server.py implements a FastMCP server exposing tools to interact with common EBS tables (e.g., OE_ORDER_HEADERS_ALL, PO_HEADERS_ALL).

`server.py`

import os
import cx_Oracle
from fastmcp import FastMCP
from typing import Optional, Dict, Any

# Ensure your container has network access (e.g. via NordLayer)

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

def get_connection():
    """
    Establishes a connection to the Oracle EBS database.
    Requires Oracle Instant Client libraries in the environment.
    """
    try:
        dsn = cx_Oracle.makedsn(
            os.getenv("ORACLE_HOST"),
            os.getenv("ORACLE_PORT", "1521"),
            service_name=os.getenv("ORACLE_SERVICE_NAME")
        )
        connection = cx_Oracle.connect(
            user=os.getenv("ORACLE_USER"),
            password=os.getenv("ORACLE_PASSWORD"),
            dsn=dsn
        )
        return connection
    except cx_Oracle.Error as e:
        raise RuntimeError(f"Oracle Connection Error: {e}")

@mcp.tool()
def get_order_status(order_number: int) -> str:
    """
    Retrieves the flow status code for a specific Sales Order in Oracle EBS.
    Target Table: OE_ORDER_HEADERS_ALL
    """
    conn = get_connection()
    cursor = conn.cursor()
    try:
        # Standard EBS query for order status
        query = """
            SELECT flow_status_code
            FROM oe_order_headers_all
            WHERE order_number = :order_id
        """
        cursor.execute(query, order_id=order_number)
        result = cursor.fetchone()

        if result:
            return f"Order {order_number} Status: {result[0]}"
        else:
            return f"Order {order_number} not found in OE_ORDER_HEADERS_ALL."
    except Exception as e:
        return f"Error querying EBS: {str(e)}"
    finally:
        cursor.close()
        conn.close()

@mcp.tool()
def check_inventory_level(item_code: str, organization_id: int) -> str:
    """
    Checks on-hand quantity for an item in a specific inventory organization.
    Target View: MTL_ONHAND_QUANTITIES_DETAIL
    """
    conn = get_connection()
    cursor = conn.cursor()
    try:
        query = """
            SELECT SUM(transaction_quantity)
            FROM mtl_onhand_quantities_detail
            WHERE inventory_item_id = (
                SELECT inventory_item_id
                FROM mtl_system_items_b
                WHERE segment1 = :item_code
                AND organization_id = :org_id
            )
            AND organization_id = :org_id
        """
        cursor.execute(query, item_code=item_code, org_id=organization_id)
        result = cursor.fetchone()

        qty = result[0] if result and result[0] is not None else 0
        return f"Item {item_code} (Org {organization_id}): {qty} units on hand."
    except Exception as e:
        return f"Error checking inventory: {str(e)}"
    finally:
        cursor.close()
        conn.close()

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

🐳 Containerization (Docker)

Using cx_Oracle requires the Oracle Instant Client libraries to be present at the OS level. Standard Python images will fail without libaio1 and the Instant Client binaries.

`Dockerfile`

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

# 1. Install system dependencies required for Oracle Instant Client
# libaio1 is critical for OCI (Oracle Call Interface)
RUN apt-get update && apt-get install -y \
    libaio1 \
    wget \
    unzip \
    && rm -rf /var/lib/apt/lists/*

# 2. Setup Oracle Instant Client (Basic Lite)
# Adjust the version number below if your EBS version requires a specific client
WORKDIR /opt/oracle
RUN wget https://download.oracle.com/otn_software/linux/instantclient/211000/instantclient-basiclite-linux.x64-21.10.0.0.0dbru.zip \
    && unzip instantclient-basiclite-linux.x64-21.10.0.0.0dbru.zip \
    && rm instantclient-basiclite-linux.x64-21.10.0.0.0dbru.zip \
    && echo /opt/oracle/instantclient_21_10 > /etc/ld.so.conf.d/oracle-instantclient.conf \
    && ldconfig

# 3. Set Environment Variables
ENV LD_LIBRARY_PATH=/opt/oracle/instantclient_21_10:$LD_LIBRARY_PATH
ENV PATH=$PATH:/opt/oracle/instantclient_21_10

# 4. Install Python Dependencies
WORKDIR /app
COPY requirements.txt .
# contents of requirements.txt:
# fastmcp
# cx_Oracle
RUN pip install --no-cache-dir -r requirements.txt

# 5. Copy Server Code
COPY server.py .

# 6. Expose Port for MCP
# EXPOSE 8000 for Railway/Cloud compatibility
EXPOSE 8000

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

🔌 Connecting to LangGraph

Once deployed (locally or on a cloud provider like Railway), you connect this MCP server to your LangGraph agent.

Environment Variables

Ensure your deployment environment has these set:

ORACLE_HOST: e.g., ebs-prod.corp.local or a VPN IP.
ORACLE_PORT: Usually 1521 or 1522.
ORACLE_SERVICE_NAME: e.g., EBSPROD.
ORACLE_USER: A service account (e.g., XX_AGENT_USER).
ORACLE_PASSWORD: The service account password.

LangGraph Implementation Example

from langgraph.prebuilt import create_react_agent
from langchain_mcp_adapters.client import Client
from langchain_openai import ChatOpenAI

# Connect to the MCP Server running via Docker
client = Client("http://localhost:8000/sse")

# Sync tools from the Oracle MCP Server
tools = await client.get_tools()

# Initialize Agent
model = ChatOpenAI(model="gpt-4-turbo")
agent = create_react_agent(model, tools)

# Run Query
response = await agent.ainvoke({
    "messages": [("user", "Check the status of Sales Order #10554 and see if we have inventory for item AS54888 in Org 204.")]
})

⚠️ Troubleshooting Legacy Errors

DPI-1047: Cannot locate a 64-bit Oracle Client library: This means LD_LIBRARY_PATH is not set correctly in the Dockerfile, or libaio1 is missing.
ORA-12170: TNS:Connect timeout: Your container cannot reach the EBS host. Check your VPN/tunnel (e.g., NordLayer, Tailscale) or firewall rules.
ORA-01017: invalid username/password: Double-check that your service account has CONNECT and SELECT permissions on the required EBS views.

🛡️ Quality Assurance

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

Transparency: This page may contain affiliate links.

Integrating LangGraph with Oracle EBS workflows using `cx_Oracle`

Integrating LangGraph with Oracle EBS workflows using cx_Oracle

🛠️ Architectural Overview

Why cx_Oracle?

💻 The Bridge Code

server.py

🐳 Containerization (Docker)

Dockerfile

🔌 Connecting to LangGraph

Environment Variables

LangGraph Implementation Example

⚠️ Troubleshooting Legacy Errors

🛡️ Quality Assurance

Integrating LangGraph with Oracle EBS workflows using `cx_Oracle`

Why `cx_Oracle`?

`server.py`

`Dockerfile`