LangGraph-orchestrated Oracle EBS automation with Node.js `oracledb`

Slug: langgraph-oracle-ebs-automation-node-js

The Retrofit Strategy

While many legacy enterprise integrations traditionally rely on Node.js middleware (using the oracledb Node driver), the rise of AI Agents often demands a Python-first architecture. Frameworks like LangGraph run natively in Python, and the Model Context Protocol (MCP) has its most robust implementation in Python via FastMCP.

This guide bridges that gap. We provide a deployment-ready Python MCP Server that functions as a direct drop-in replacement for what might otherwise be a Node.js microservice. By using python-oracledb—the sibling to the Node driver—we maintain the high-performance connectivity you expect while enabling native “plug-and-play” compatibility with LangGraph agents.

Why this approach?

Native Compatibility: LangGraph agents can talk directly to this server without complex JSON bridging.
Thin Client Mode: We utilize python-oracledb in “Thin” mode, eliminating the need for bulky Oracle Instant Client installations in Docker.
Unified Stack: Keep your agent logic and connectivity logic in the same language ecosystem for easier debugging.

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

This server implements the Model Context Protocol (MCP) using FastMCP. It exposes specific Oracle EBS functions (like checking inventory or invoice status) as executable tools for your AI agent.

import os
import oracledb
from fastmcp import FastMCP

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

def get_connection():
    """Establishes a connection to the Oracle EBS database."""
    # Ensure your container has network access (e.g. via NordLayer)
    try:
        connection = oracledb.connect(
            user=os.getenv("ORACLE_USER"),
            password=os.getenv("ORACLE_PASSWORD"),
            dsn=os.getenv("ORACLE_DSN")  # e.g., "dbhost.example.com:1521/ebspdb1"
        )
        return connection
    except oracledb.Error as e:
        raise RuntimeError(f"Oracle Connection Failed: {e}")

@mcp.tool()
def check_invoice_status(invoice_num: str) -> str:
    """
    Queries Oracle EBS (AP_INVOICES_ALL) to get the status of a specific invoice.
    Args:
        invoice_num: The invoice number to lookup (e.g., 'INV-2023-001').
    """
    conn = None
    cursor = None
    try:
        conn = get_connection()
        cursor = conn.cursor()

        # Standard Oracle EBS query structure
        query = """
            SELECT invoice_num, invoice_amount, payment_status_flag, status_lookup_code
            FROM ap.ap_invoices_all
            WHERE invoice_num = :inv_num
        """

        cursor.execute(query, inv_num=invoice_num)
        row = cursor.fetchone()

        if row:
            return (f"Invoice {row[0]}: Amount=${row[1]}, "
                    f"Payment Status={row[2]}, Status={row[3]}")
        else:
            return f"Invoice {invoice_num} not found in EBS."

    except Exception as e:
        return f"Error querying invoice: {str(e)}"
    finally:
        if cursor:
            cursor.close()
        if conn:
            conn.close()

@mcp.tool()
def get_inventory_level(item_sku: str, organization_id: int) -> str:
    """
    Checks on-hand quantity for an item in a specific EBS Organization.
    Args:
        item_sku: The part number or SKU.
        organization_id: The Oracle Inventory Organization ID.
    """
    conn = None
    cursor = None
    try:
        conn = get_connection()
        cursor = conn.cursor()

        # Querying MTL_ONHAND_QUANTITIES for stock levels
        query = """
            SELECT SUM(transaction_quantity)
            FROM inv.mtl_onhand_quantities
            WHERE inventory_item_id = (
                SELECT inventory_item_id
                FROM inv.mtl_system_items_b
                WHERE segment1 = :sku
                AND organization_id = :org_id
            )
            AND organization_id = :org_id
        """

        cursor.execute(query, sku=item_sku, org_id=organization_id)
        result = cursor.fetchone()

        if result and result[0] is not None:
            return f"Current on-hand quantity for {item_sku}: {result[0]}"
        else:
            return f"Item {item_sku} not found or no stock in Org {organization_id}."

    except Exception as e:
        return f"Error checking inventory: {str(e)}"
    finally:
        if cursor:
            cursor.close()
        if conn:
            conn.close()

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

2. Containerization (`Dockerfile`)

We use a slim Python image. Because python-oracledb runs in Thin Mode by default, we do not need to install the heavy Oracle Instant Client libraries unless you are using very old legacy encryption protocols.

Note: Railway requires EXPOSE 8000 (or PORT env var) to route traffic correctly.

# Use an official Python runtime as a parent image
FROM python:3.11-slim

# Set work directory
WORKDIR /app

# Install system dependencies (minimal)
# Ensure your container has network access (e.g. via NordLayer)
RUN apt-get update && apt-get install -y \
    gcc \
    libaio1 \
    && rm -rf /var/lib/apt/lists/*

# Copy requirements or install directly
# fastmcp handles the server, oracledb handles the database
RUN pip install --no-cache-dir fastmcp oracledb

# Copy the server code
COPY server.py .

# Expose the port for Railway/Cloud Run
EXPOSE 8000

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

3. LangGraph Integration

Once your MCP server is running (e.g., on http://localhost:8000 or a Railway URL), you can connect it to your LangGraph agent.

# Example LangGraph Node configuration (Conceptual)
from langgraph.prebuilt import ToolNode
from mcp import ClientSession

# This represents the tool definition your agent receives
tools = [
    {
        "type": "function",
        "function": {
            "name": "check_invoice_status",
            "description": "Check AP invoice status in Oracle EBS",
            "parameters": { ... }
        }
    }
]

# The agent can now emit tool calls that map directly
# to your running MCP server's endpoints.

Common Troubleshooting

ORA-12154: TNS:could not resolve the connect identifier:
- Cause: The ORACLE_DSN format is incorrect or the container cannot resolve the hostname.
- Fix: Use the “Easy Connect” syntax for DSN: hostname:port/service_name (e.g., oracle-prod.corp:1521/EBSPROD). Avoid tnsnames.ora dependencies in containers if possible.
Connection Timeout:
- Cause: Oracle EBS is usually behind a corporate firewall.
- Fix: Ensure your Docker container is running within a VPN context (like NordLayer or Tailscale sidecar) if the database is on-prem.

🛡️ Quality Assurance

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

Transparency: This page may contain affiliate links.

LangGraph-orchestrated Oracle EBS automation with Node.js `oracledb`

LangGraph-orchestrated Oracle EBS automation with Node.js oracledb

The Retrofit Strategy

Why this approach?

1. The Protocol Server (server.py)

2. Containerization (Dockerfile)

3. LangGraph Integration

Common Troubleshooting

🛡️ Quality Assurance

LangGraph-orchestrated Oracle EBS automation with Node.js `oracledb`

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

2. Containerization (`Dockerfile`)