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

Connecting autonomous AI Agents to Oracle E-Business Suite (EBS) is rarely a “happy path” operation. Legacy databases are notoriously sensitive to connection spikes, timeouts, and locking contention. When an Agent (like CrewAI or AutoGen) attempts to insert a Purchase Order or query inventory, a single ORA-12170 can cause the entire agentic workflow to hallucinate or crash if not handled gracefully.

This guide provides a production-ready Model Context Protocol (MCP) server that wraps Oracle EBS interactions in a “Circuit Breaker” pattern. It ensures your agents receive structured, actionable error messages instead of raw stack traces, allowing them to self-correct or escalate.

Architectural Pattern

We use FastMCP to create a lightweight middleware server. This server:

Intercepts Agent requests.
Attempts the Oracle EBS operation using oracledb (the modern thin client).
Catches specific ORA codes (Connection, Auth, Integrity).
Retries transient errors (like network blips) automatically.
Returns a clean JSON string to the Agent, describing the success or the specific failure reason.

The “Safe-Fail” Protocol

An Agent should never see a raw Python traceback. Instead, we return responses like:

{"status": "error", "code": "ORA-00054", "message": "Record locked by another user. Retry in 5 minutes."}
{"status": "success", "data": [...]}

The Code

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

This server implements a decorator-based retry mechanism and exposes tools for querying EBS.

import os
import time
import functools
import oracledb
from fastmcp import FastMCP

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

# --- Configuration ---
# Ensure your container has network access (e.g. via NordLayer) to reach the on-prem Oracle DB.
DB_USER = os.getenv("ORACLE_USER", "APPS")
DB_PASSWORD = os.getenv("ORACLE_PASSWORD", "welcome1")
DB_DSN = os.getenv("ORACLE_DSN", "ebs_prod_host:1521/VIS")

def get_connection():
    """Establishes a connection to Oracle EBS."""
    return oracledb.connect(
        user=DB_USER,
        password=DB_PASSWORD,
        dsn=DB_DSN
    )

def oracle_error_handler(func):
    """
    Decorator to wrap Oracle interactions in robust error handling.
    Translates cryptic ORA codes into Agent-readable messages.
    """
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        retries = 3
        delay = 2

        for attempt in range(retries):
            try:
                return func(*args, **kwargs)

            except oracledb.OperationalError as e:
                error_obj, = e.args
                # Handle Transient Network/Connection Errors
                if error_obj.code in [12170, 12541, 3113, 3114]:
                    if attempt < retries - 1:
                        time.sleep(delay)
                        delay *= 2  # Exponential backoff
                        continue
                    return f"Error: Network unreachable after {retries} attempts. (ORA-{error_obj.code})"
                else:
                    return f"Error: Operational DB failure. {error_obj.message}"

            except oracledb.IntegrityError as e:
                error_obj, = e.args
                # Handle Data Constraints (Duplicate keys, FK violations)
                return f"Error: Data integrity violation. Please check your input parameters. (ORA-{error_obj.code})"

            except oracledb.DatabaseError as e:
                error_obj, = e.args
                # Handle Invalid Credentials or Permissions
                if error_obj.code == 1017:
                    return "Error: Invalid Credentials. Please verify username/password."
                return f"Error: Database exception: {error_obj.message}"

            except Exception as e:
                return f"Error: Unexpected system failure: {str(e)}"

    return wrapper

@mcp.tool()
@oracle_error_handler
def query_supplier_status(supplier_name: str) -> str:
    """
    Queries Oracle EBS for a supplier's status.
    Use this to check if a vendor is active before creating invoices.
    """
    query = """
        SELECT vendor_name, enabled_flag, segment1
        FROM ap_suppliers
        WHERE vendor_name = :s_name
    """

    with get_connection() as conn:
        with conn.cursor() as cursor:
            cursor.execute(query, s_name=supplier_name)
            row = cursor.fetchone()

            if row:
                return f"Supplier Found: Name={row[0]}, Active={row[1]}, ID={row[2]}"
            else:
                return "Status: Supplier not found in AP_SUPPLIERS table."

@mcp.tool()
@oracle_error_handler
def check_invoice_status(invoice_num: str) -> str:
    """
    Checks the status of a specific invoice in AP_INVOICES_ALL.
    """
    query = """
        SELECT invoice_num, invoice_amount, payment_status_flag
        FROM ap_invoices_all
        WHERE invoice_num = :i_num
    """

    with get_connection() as conn:
        with conn.cursor() as cursor:
            cursor.execute(query, i_num=invoice_num)
            row = cursor.fetchone()

            if row:
                return f"Invoice {row[0]}: Amount=${row[1]}, Paid={row[2]}"
            else:
                return f"Invoice {invoice_num} does not exist."

if __name__ == "__main__":
    # HOST must be 0.0.0.0 to work in Docker
    mcp.run(transport='sse', host='0.0.0.0', port=8000)

2. The Dockerfile

We need a Python environment with oracledb. This modern library does not require the heavy Oracle Instant Client for standard TCP connections (Thin Mode), making the Docker image significantly smaller (approx 200MB vs 1GB+).

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

# Set working directory
WORKDIR /app

# Install dependencies
# oracledb is the modern replacement for cx_Oracle
RUN pip install --no-cache-dir \
    fastmcp \
    oracledb \
    uvicorn

# Copy server code
COPY server.py .

# Ensure standard output is logged immediately
ENV PYTHONUNBUFFERED=1

# Expose port 8000 for Railway/Docker networking
EXPOSE 8000

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

Connecting the Agent (Client Side)

Once your Docker container is running, you can connect a CrewAI agent to it. We use the mcps configuration parameter to automatically ingest the tools exposed by our server.

Client Code (CrewAI)

from crewai import Agent, Task, Crew

# 1. Define the Agent with MCP connectivity
# The agent will automatically discover tools from the MCP server.
oracle_analyst = Agent(
    role='Oracle EBS Analyst',
    goal='Check supplier status and validate invoices in legacy Oracle systems',
    backstory="""You are a senior ERP specialist. You are careful.
    If you receive a network error, you wait and retry.
    If you receive a 'Supplier not found' error, you ask the user for the correct spelling.""",
    mcps=["http://localhost:8000/sse"]  # Connects to the Docker container
)

# 2. Define the Task
# The agent will attempt to call 'query_supplier_status' provided by the MCP server
check_supplier_task = Task(
    description="Check if 'Acme Corp' is an active supplier. If the database is down, retry once.",
    expected_output="A summary of the supplier status or a clear error report.",
    agent=oracle_analyst
)

# 3. Run the Crew
crew = Crew(
    agents=[oracle_analyst],
    tasks=[check_supplier_task],
    verbose=True
)

result = crew.kickoff()
print("Final Result:", result)

How It Works in Practice

Scenario A: Network Fluctuation

Agent Action: Calls query_supplier_status("Acme Corp").
Server Event: Oracle DB is temporarily unreachable (ORA-12170).
Middleware Logic: The oracle_error_handler catches the error. It sleeps for 2 seconds, then 4 seconds.
Recovery: The 3rd attempt succeeds.
Agent Receipt: The Agent receives Supplier Found: Name=Acme Corp... without ever knowing a failure occurred.

Scenario B: Bad Data

Agent Action: Calls query_supplier_status with a name that triggers a character set error or SQL injection attempt.
Server Event: oracledb.DatabaseError occurs.
Middleware Logic: The handler immediately catches it and returns Error: Database exception: ....
Agent Receipt: The Agent sees the string starting with “Error:”.
Agent Reaction: Based on its backstory, the Agent concludes: “I encountered a database error. I cannot proceed with this specific query.”

Testing Your Implementation

Build the Container:
Terminal window
```
docker build -t oracle-mcp-agent .
```

Run with Environment Variables:

docker run -p 8000:8000 \
  -e ORACLE_USER=APPS \
  -e ORACLE_PASSWORD=manager \
  -e ORACLE_DSN=192.168.1.50:1521/VIS \
  oracle-mcp-agent

Verify: Run the Client Code script. The CrewAI agent will connect to http://localhost:8000/sse, discover the query_supplier_status tool, and execute the task.

🛡️ 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)

Architectural Pattern

The “Safe-Fail” Protocol

The Code

1. The MCP Server (server.py)

2. The Dockerfile

Connecting the Agent (Client Side)

Client Code (CrewAI)

How It Works in Practice

Scenario A: Network Fluctuation

Scenario B: Bad Data

Testing Your Implementation

🛡️ Quality Assurance

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