Robust Error Handling for AI Agents in Oracle EBS integrations (Python)

When AI agents like CrewAI, LangGraph, or OpenAI Operator interact with Oracle E-Business Suite (EBS), they often encounter the cryptic “ORA-XXXXX” error codes typical of databases designed in the 1990s.

A standard LLM does not inherently know that ORA-12170 means “check the VPN” while ORA-02292 means “you can’t delete this because a child record exists.” Without an abstraction layer, an agent will hallucinate a fix or loop endlessly, retrying the same doomed query.

This guide provides a production-ready FastMCP server designed to wrap Oracle EBS interactions. It catches raw Oracle exceptions, interprets them, and returns semantic, actionable instructions that allow your AI agent to self-correct.

The Challenge: Why Agents Fail with Oracle

Legacy ERPs like Oracle EBS are unforgiving. A slight schema mismatch or a network blip results in a hard crash.

Error Code	Raw Oracle Message	What it means for the Agent
ORA-12170	`TNS:Connect timeout occurred`	Network Failure. The agent should check if the VPN/Tunnel is active, not retry immediately.
ORA-00942	`table or view does not exist`	Hallucination. The agent guessed a table name (e.g., `USERS` instead of `FND_USER`). It needs to ask for the schema schema.
ORA-01017	`invalid username/password`	Auth Failure. The credentials in `.env` are wrong or expired. Retrying won’t help.
ORA-02292	`integrity constraint violated - child record found`	Business Logic. The agent is trying to delete a parent record (like a PO Header) without removing lines first.

The Solution: A Semantic Error Wrapper

We will build a FastMCP tool that acts as a “Smart Gateway.” Instead of crashing, it catches oracledb.Error, parses the code, and returns a JSON object with a status: "error" and a suggested_action field tailored for an LLM.

Prerequisites

Python 3.10+
Oracle Database Access: You need the host, service name, and credentials.
Network Access: Your Docker container must be able to reach the Oracle host (often requires a VPN or Private Link).

1. The Code (`server.py`)

This server uses oracledb (the modern, thin-client successor to cx_Oracle) and fastmcp.

import os
import oracledb
import json
from fastmcp import FastMCP

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

# Ensure your container has network access (e.g. via NordLayer)
# Configuration via Environment Variables
DB_USER = os.getenv("ORACLE_USER", "APPS")
DB_PASSWORD = os.getenv("ORACLE_PASSWORD", "welcome1")
DB_DSN = os.getenv("ORACLE_DSN", "ebs.example.com:1521/VIS")

def get_connection():
    """Establishes a connection to the Oracle EBS database."""
    try:
        # standard thin client connection
        connection = oracledb.connect(
            user=DB_USER,
            password=DB_PASSWORD,
            dsn=DB_DSN
        )
        return connection
    except oracledb.Error as e:
        raise e

def semantic_error_parser(error_obj):
    """
    Translates cryptic ORA codes into agent-friendly instructions.
    """
    error_code = error_obj.code
    message = error_obj.message

    # Default response
    response = {
        "status": "error",
        "code": f"ORA-{error_code}",
        "raw_message": message,
        "classification": "UNKNOWN",
        "instruction": "Read the raw message and attempt to interpret."
    }

    if error_code == 12170 or error_code == 12541:
        response["classification"] = "NETWORK_TIMEOUT"
        response["instruction"] = "The database is unreachable. 1) Check VPN connection. 2) Verify hostname/port in config. Do NOT retry immediately."

    elif error_code == 1017:
        response["classification"] = "AUTHENTICATION_FAILURE"
        response["instruction"] = "Invalid credentials. Stop. Ask the human administrator to verify ORACLE_USER and ORACLE_PASSWORD."

    elif error_code == 942:
        response["classification"] = "SCHEMA_ERROR"
        response["instruction"] = "The table does not exist. You may be hallucinating table names. Use a tool to list available tables (e.g., SELECT table_name FROM all_tables) before retrying."

    elif error_code == 904:
        response["classification"] = "SCHEMA_ERROR"
        response["instruction"] = "Invalid identifier (column name). Check the table definition."

    elif error_code == 2292:
        response["classification"] = "DATA_INTEGRITY"
        response["instruction"] = "You cannot delete this record because child records exist. Delete the child records first."

    elif error_code == 1400:
        response["classification"] = "DATA_VALIDATION"
        response["instruction"] = "You attempted to insert NULL into a mandatory column. Check the schema for NOT NULL constraints."

    return response

@mcp.tool()
def execute_ebs_query(sql_query: str, params: dict = None) -> str:
    """
    Safely executes a SQL query against Oracle EBS with robust error handling.

    Args:
        sql_query: The SQL statement to execute (e.g., "SELECT * FROM FND_USER WHERE USER_NAME = :un").
        params: Dictionary of bind variables (e.g., {"un": "SYSADMIN"}).

    Returns:
        JSON string containing the results or a semantic error analysis.
    """
    connection = None
    cursor = None
    try:
        connection = get_connection()
        cursor = connection.cursor()

        # Oracle bind variables usually require strict typing, oracledb handles basic types well.
        # Defaults to empty dict if None
        safe_params = params if params else {}

        cursor.execute(sql_query, safe_params)

        # If it's a SELECT query, fetch results
        if sql_query.strip().upper().startswith("SELECT"):
            columns = [col[0] for col in cursor.description]
            rows = cursor.fetchall()
            results = [dict(zip(columns, row)) for row in rows]
            return json.dumps({
                "status": "success",
                "row_count": len(results),
                "data": results
            }, default=str)
        else:
            # For INSERT/UPDATE/DELETE, commit the transaction
            connection.commit()
            return json.dumps({
                "status": "success",
                "message": "Statement executed and committed successfully."
            })

    except oracledb.Error as e:
        # Intercept the Oracle error and return semantic JSON
        error_context = semantic_error_parser(e)
        return json.dumps(error_context)

    except Exception as e:
        # Catch generic python errors
        return json.dumps({
            "status": "error",
            "classification": "SYSTEM_ERROR",
            "message": str(e)
        })

    finally:
        if cursor:
            cursor.close()
        if connection:
            connection.close()

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

2. The Container (`Dockerfile`)

We use a slim Python image. Note the EXPOSE 8000 instruction, which is critical for hosting this on platforms like Railway or Render where your Agent might look for the API.

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

# Set the working directory in the container
WORKDIR /app

# Install system dependencies if needed (usually none for oracledb thin mode)
# If using Thick mode (for advanced wallet auth), you would install libaio1 here.

# Copy the current directory contents into the container at /app
COPY server.py .

# Install any needed packages
# fastmcp: The MCP server framework
# oracledb: The official Oracle driver (Thin mode by default)
RUN pip install --no-cache-dir fastmcp oracledb

# Make port 8000 available to the world outside this container
EXPOSE 8000

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

3. Usage Guide

Building and Running

Build the image:
Terminal window
```
docker build -t oracle-ebs-gateway .
```

Run with Environment Variables:

docker run -p 8000:8000 \
  -e ORACLE_USER="APPS" \
  -e ORACLE_PASSWORD="YOUR_PASSWORD" \
  -e ORACLE_DSN="ebs.company.net:1521/PROD" \
  oracle-ebs-gateway

Testing with an Agent

When you connect this MCP server to Claude Desktop or an OpenAI Agent, the agent can now “reason” about errors.

Scenario: The agent attempts to query a non-existent table PO_HEADERS (the real table is PO_HEADERS_ALL).

Agent Request: execute_ebs_query("SELECT * FROM PO_HEADERS WHERE ROWNUM < 5")

Response (JSON):

{
  "status": "error",
  "code": "ORA-942",
  "classification": "SCHEMA_ERROR",
  "instruction": "The table does not exist. You may be hallucinating table names. Use a tool to list available tables..."
}

Agent Reaction: “Ah, I made a schema error. I will list the tables matching ‘PO_HEADERS’ to find the correct name.”

This loop turns a fatal crash into a successful, autonomous debugging step.

🛡️ Quality Assurance

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

Transparency: This page may contain affiliate links.

Robust Error Handling for AI Agents in Oracle EBS integrations (Python)

Robust Error Handling for AI Agents in Oracle EBS integrations (Python)

The Challenge: Why Agents Fail with Oracle

The Solution: A Semantic Error Wrapper

Prerequisites

1. The Code (server.py)

2. The Container (Dockerfile)

3. Usage Guide

Building and Running

Testing with an Agent

🛡️ Quality Assurance

1. The Code (`server.py`)

2. The Container (`Dockerfile`)