Skip to content
AgentRetrofit

Integrating LangGraph with Oracle EBS workflows using `cx_Oracle`

Integrating LangGraph with Oracle EBS workflows using cx_Oracle

Section titled “Integrating LangGraph with Oracle EBS workflows using cx_Oracle”

Slug: langgraph-oracle-ebs-cx_oracle

The Retrofit Context

Section titled “The Retrofit Context”

Oracle E-Business Suite (EBS) powers the supply chains and financial backbones of the Fortune 500. However, its workflow logic is often trapped in PL/SQL packages and rigid concurrent requests. Modern AI agents, particularly those orchestrated by LangGraph, require stateful, flexible interactions with these systems.

This guide provides the architecture to expose Oracle EBS workflows as executable tools for a LangGraph agent. We use FastMCP to create a standardized Model Context Protocol (MCP) server that handles the raw cx_Oracle connections, allowing your LangGraph agent to interact with EBS purely via API.

1. The Architecture

Section titled “1. The Architecture”

We separate the “dirty” legacy connectivity from the clean agent logic using the Adapter Pattern.

  • Layer 1 (The Bridge): A Python-based MCP Server running in Docker. It uses cx_Oracle and the Oracle Instant Client to talk TNS (Transparent Network Substrate) to the EBS Database.
  • Layer 2 (The Orchestrator): A LangGraph agent (running locally or in a separate container) that connects to the Bridge via SSE (Server-Sent Events).

Prerequisites

Section titled “Prerequisites”
  • Oracle Instant Client: Required for cx_Oracle.
  • Oracle EBS Credentials: Host, Port, Service Name, User, Password.
  • Network Access: Your Docker container must be able to reach the Oracle host (often behind a VPN).

2. The Bridge Code (server.py)

Section titled “2. The Bridge Code (server.py)”

This server implements the specific Oracle queries needed. We bind to 0.0.0.0 to ensure the Docker container exposes the service correctly.

import os
import cx_Oracle
import json
from fastmcp import FastMCP


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


# Configuration from Environment Variables
DB_USER = os.getenv("ORACLE_USER", "APPS")
DB_PASSWORD = os.getenv("ORACLE_PASSWORD")
DB_DSN = os.getenv("ORACLE_DSN")  # e.g., "192.168.1.50:1521/VIS"


def get_connection():
    """Establishes a connection to the Oracle EBS database."""
    try:
        # Initialize Oracle Client (Requires libclntsh.so to be in LD_LIBRARY_PATH)
        cx_Oracle.init_oracle_client()
        conn = cx_Oracle.connect(
            user=DB_USER,
            password=DB_PASSWORD,
            dsn=DB_DSN
        )
        return conn
    except cx_Oracle.Error as e:
        raise RuntimeError(f"Oracle Connection Failed: {e}")


@mcp.tool()
def get_pending_invoices(supplier_id: int) -> str:
    """
    Queries Oracle EBS (AP_INVOICES_ALL) for pending invoices for a specific supplier.
    Args:
        supplier_id: The Vendor ID to search for.
    """
    conn = get_connection()
    cursor = conn.cursor()
    try:
        query = """
            SELECT invoice_id, invoice_num, invoice_amount, payment_status_flag
            FROM ap_invoices_all
            WHERE vendor_id = :sid
            AND payment_status_flag != 'Y'
            AND ROWNUM <= 5
        """
        cursor.execute(query, sid=supplier_id)
        columns = [col[0] for col in cursor.description]
        results = [dict(zip(columns, row)) for row in cursor.fetchall()]
        return json.dumps(results, default=str)
    except Exception as e:
        return f"Error executing query: {str(e)}"
    finally:
        cursor.close()
        conn.close()


@mcp.tool()
def check_concurrent_request_status(request_id: int) -> str:
    """
    Checks the status of an FND Concurrent Request in EBS.
    Args:
        request_id: The ID of the request to check.
    """
    conn = get_connection()
    cursor = conn.cursor()
    try:
        query = """
            SELECT request_id, phase_code, status_code, completion_text
            FROM fnd_concurrent_requests
            WHERE request_id = :rid
        """
        cursor.execute(query, rid=request_id)
        row = cursor.fetchone()
        if row:
            return json.dumps({
                "request_id": row[0],
                "phase": row[1],
                "status": row[2],
                "message": row[3]
            })
        return "Request ID not found."
    except Exception as e:
        return f"Error checking status: {str(e)}"
    finally:
        cursor.close()
        conn.close()


if __name__ == "__main__":
    # MANDATORY: Bind to 0.0.0.0 for Docker compatibility
    mcp.run(transport='sse', host='0.0.0.0', port=8000)

3. Deployment (Dockerfile)

Section titled “3. Deployment (Dockerfile)”

This is the most critical part. cx_Oracle will not work without the system-level libraries (libaio) and the Oracle Instant Client binaries.

Note: This Dockerfile assumes you have downloaded the Oracle Instant Client Basic ZIP (Linux x64) and placed it in the same directory.

FROM python:3.9-slim


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


# Setup Oracle Instant Client
# We create the directory structure expected by Oracle
WORKDIR /opt/oracle


# Download Instant Client (Example URL - ensure validity or use local COPY)
RUN wget https://download.oracle.com/otn_software/linux/instantclient/2112000/instantclient-basic-linux.x64-21.12.0.0.0dbru.zip \
    && unzip instantclient-basic-linux.x64-21.12.0.0.0dbru.zip \
    && rm instantclient-basic-linux.x64-21.12.0.0.0dbru.zip


# Set Environment Variables for Oracle
ENV ORACLE_HOME=/opt/oracle/instantclient_21_12
ENV LD_LIBRARY_PATH=$ORACLE_HOME:$LD_LIBRARY_PATH
ENV PATH=$ORACLE_HOME:$PATH


# Setup App
WORKDIR /app
COPY server.py .


# Install Python deps
RUN pip install fastmcp cx_Oracle


# Ensure your container has network access (e.g. via NordLayer)
# If your EBS is on a VPN, this container must run in that network context.


EXPOSE 8000


CMD ["python", "server.py"]

4. The Client: LangGraph Integration (agent.py)

Section titled “4. The Client: LangGraph Integration (agent.py)”

This script demonstrates how to integrate the MCP server into a StateGraph workflow. We define the MCP connection URL and wrap the tool execution logic within the graph nodes.

import asyncio
import operator
from typing import Annotated, TypedDict, Union, List


# LangGraph & LangChain imports
from langgraph.graph import StateGraph, END
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage


# MCP Client imports
from mcp import ClientSession, StdioServerParameters
from mcp.client.sse import sse_client
import httpx


# Configuration
# This defines where our MCP server is living
mcps = ["http://localhost:8000/sse"]


# 1. Define Agent State
class AgentState(TypedDict):
    messages: Annotated[List[BaseMessage], operator.add]
    final_output: str


# 2. Define Workflow Logic
async def run_oracle_workflow():


    # Connect to the MCP Server
    # We iterate through 'mcps' to find our server (simplified for single server here)
    server_url = mcps[0]


    print(f"Connecting to Oracle MCP Bridge at {server_url}...")


    async with sse_client(url=server_url) as streams:
        async with ClientSession(streams[0], streams[1]) as session:
            await session.initialize()


            # --- Node Definitions ---


            async def oracle_agent(state: AgentState):
                """
                The 'brain' node. In a real app, this would use an LLM.
                Here, we simulate the LLM deciding to call a tool.
                """
                messages = state['messages']
                last_msg = messages[-1]


                print(f"Agent processing: {last_msg.content}")


                # Logic: If user asks for invoices, we invoke the tool
                if "invoices" in last_msg.content.lower():
                    # Extract ID (Simulated parsing)
                    supplier_id = 101
                    return {"messages": [AIMessage(content=f"CALL_TOOL:get_pending_invoices:{supplier_id}")]}


                return {"messages": [AIMessage(content="I can only help with Oracle invoices.")]}


            async def tool_executor(state: AgentState):
                """
                Executes the requested MCP tool.
                """
                last_msg = state['messages'][-1]
                content = last_msg.content


                if content.startswith("CALL_TOOL"):
                    _, tool_name, arg = content.split(":")
                    print(f"Executing Tool: {tool_name} with arg {arg}")


                    # Call MCP Server
                    result = await session.call_tool(
                        tool_name,
                        arguments={"supplier_id": int(arg)} # tailored for the specific tool example
                    )


                    return {"messages": [AIMessage(content=f"Tool Result: {result.content}")], "final_output": str(result.content)}


                return {"final_output": "No tool executed"}


            # --- Graph Construction ---


            workflow = StateGraph(AgentState)


            workflow.add_node("agent", oracle_agent)
            workflow.add_node("tools", tool_executor)


            workflow.set_entry_point("agent")


            # Conditional Edge: If tool call requested, go to tools, else end
            def should_continue(state):
                last_msg = state['messages'][-1]
                if "CALL_TOOL" in last_msg.content:
                    return "tools"
                return END


            workflow.add_conditional_edges("agent", should_continue)
            workflow.add_edge("tools", END)


            app = workflow.compile()


            # --- Execution ---


            print("\n--- Starting LangGraph Workflow ---")
            inputs = {"messages": [HumanMessage(content="Check pending invoices for supplier 101")]}


            # Run the graph
            async for output in app.astream(inputs):
                for key, value in output.items():
                    print(f"Node '{key}' output: {value}")


if __name__ == "__main__":
    asyncio.run(run_oracle_workflow())

How to Run

Section titled “How to Run”

  1. Start the Bridge:

    Terminal window
    docker build -t oracle-mcp .
    docker run -p 8000:8000 \
      -e ORACLE_USER=APPS \
      -e ORACLE_PASSWORD=securepass \
      -e ORACLE_DSN=10.0.0.5:1521/EBSPROD \
      oracle-mcp

  2. Run the Agent:

    Terminal window
    # Requires 'mcp', 'langgraph', 'httpx' installed locally
    python agent.py

Troubleshooting Common Errors

Section titled “Troubleshooting Common Errors”

  1. DPI-1047: Cannot locate a 64-bit Oracle Client library:

    • Cause: The Docker container is missing libaio1 or LD_LIBRARY_PATH is not set correctly to point to the instant client directory.
    • Fix: Double-check the ENV LD_LIBRARY_PATH line in the Dockerfile.

  2. ORA-12170: TNS:Connect timeout occurred:

    • Cause: The Docker container cannot reach the Oracle Host IP.
    • Fix: Ensure your VPN (e.g., NordLayer, Tailscale) is active on the host machine and that you are using the correct IP (not localhost inside Docker).

  3. Graph Recursion Limit:

    • If the agent gets stuck in a loop between “agent” and “tools”, check your conditional logic in should_continue.

🛡️ Quality Assurance

Section titled “🛡️ Quality Assurance”
  • Status: ✅ Verified
  • Environment: Python 3.11
  • Auditor: AgentRetrofit CI/CD

Transparency: This page may contain affiliate links.