AutoGen integration with Oracle EBS for workflow automation

AutoGen Integration with Oracle EBS for Workflow Automation

This guide provides a “Retrofit” solution for connecting Microsoft AutoGen agents to Oracle E-Business Suite (EBS) using the Model Context Protocol (MCP). It allows autonomous agents to trigger legacy PL/SQL workflows (Concurrent Requests) via a standardized, Dockerized interface.

🏗️ Architectural Overview

Server (FastMCP): A Python container running oracledb that exposes EBS workflows as MCP tools.
Client (AutoGen): An AutoGen workflow configured with a list of MCP server endpoints (mcps) to dynamically load tools.

🚀 Step 1: The MCP Server (`server.py`)

This server wraps the Oracle FND_REQUEST package, allowing agents to submit jobs and check their status without direct SQL knowledge.

import os
import oracledb
from mcp.server.fastmcp import FastMCP
from pydantic import Field

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

# Database Connection Configuration
DB_USER = os.getenv("EBS_DB_USER", "APPS")
DB_PASSWORD = os.getenv("EBS_DB_PASSWORD", "APPS")
DB_DSN = os.getenv("EBS_DB_DSN", "ebs.example.com:1521/VIS")

def get_connection():
    """Establishes a connection to the Oracle EBS database."""
    # Ensure your container has network access (e.g. via NordLayer)
    return oracledb.connect(
        user=DB_USER,
        password=DB_PASSWORD,
        dsn=DB_DSN
    )

@mcp.tool()
def submit_ebs_request(
    program: str = Field(..., description="Short name of the concurrent program (e.g., 'OEXOEORD')"),
    application: str = Field(..., description="Short name of the application (e.g., 'ONT')"),
    description: str = Field(None, description="Description for the request")
) -> str:
    """
    Submits a standard Oracle EBS Concurrent Request (Workflow) via PL/SQL.
    """
    conn = get_connection()
    cursor = conn.cursor()
    try:
        # FND_REQUEST.SUBMIT_REQUEST wrapper
        cursor.callfunc(
            "FND_REQUEST.SUBMIT_REQUEST",
            int,
            [application, program, description, None, False]
        )
        conn.commit()
        return f"Request submitted for Program {program} ({application})."
    except oracledb.Error as e:
        return f"Oracle EBS Error: {str(e)}"
    finally:
        cursor.close()
        conn.close()

@mcp.tool()
def check_request_status(request_id: int) -> str:
    """
    Checks the status of a specific Concurrent Request in FND_CONCURRENT_REQUESTS.
    """
    conn = get_connection()
    cursor = conn.cursor()
    try:
        sql = "SELECT phase_code, status_code FROM fnd_concurrent_requests WHERE request_id = :rid"
        cursor.execute(sql, rid=request_id)
        row = cursor.fetchone()
        if row:
            return f"Request {request_id}: Phase={row[0]}, Status={row[1]}"
        return f"Request {request_id} not found."
    except oracledb.Error as e:
        return f"Database Error: {str(e)}"
    finally:
        cursor.close()
        conn.close()

if __name__ == "__main__":
    # Binds to 0.0.0.0 to allow external Docker access
    mcp.run(transport='sse', host='0.0.0.0', port=8000)

🐳 Step 2: Dockerfile

This Dockerfile sets up the environment and exposes port 8000 for the SSE transport.

FROM python:3.11-slim

WORKDIR /app

# Install system dependencies
RUN apt-get update && apt-get install -y libaio1 && rm -rf /var/lib/apt/lists/*

# Install Python dependencies
RUN pip install --no-cache-dir mcp oracledb

# Copy server code
COPY server.py .

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

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

🔌 Step 3: Connecting AutoGen

We use a connect_mcp_tools helper to iterate through the mcps configuration list, establishing sessions and registering tools with the AutoGen agents.

import asyncio
from autogen import AssistantAgent, UserProxyAgent
from mcp import ClientSession, StdioServerParameters
from mcp.client.sse import sse_client

# Configuration: List of MCP Servers
mcps = ["http://localhost:8000/sse"]

async def connect_mcp_tools(agent: AssistantAgent, proxy: UserProxyAgent, mcp_urls: list):
    """
    Connects to a list of MCP servers and registers their tools with AutoGen agents.
    """
    sessions = []

    for url in mcp_urls:
        # Create the client context manager (needs to be kept alive in a real app)
        # For this script, we enter the context here
        client = sse_client(url)
        streams = await client.__aenter__()
        session = ClientSession(streams[0], streams[1])
        await session.__aenter__()
        await session.initialize()

        # List and register tools
        tools_list = await session.list_tools()
        print(f"Connected to {url}, found tools: {[t.name for t in tools_list.tools]}")

        for tool in tools_list.tools:
            # Create a closure to capture the specific session and tool name
            async def tool_wrapper(**kwargs):
                result = await session.call_tool(tool.name, arguments=kwargs)
                return result.content[0].text

            # Register with LLM (Assistant) and Executor (UserProxy)
            agent.register_for_llm(name=tool.name, description=tool.description)(tool_wrapper)
            proxy.register_for_execution(name=tool.name)(tool_wrapper)

        sessions.append((client, session))

    return sessions

async def run_workflow():
    # 1. Initialize Agents
    config_list = [{"model": "gpt-4", "api_key": "YOUR_OPENAI_API_KEY"}]

    ebs_admin = AssistantAgent(
        name="OracleAdmin",
        system_message="You are an Oracle EBS Admin. Use provided tools to manage concurrent requests.",
        llm_config={"config_list": config_list}
    )

    user_proxy = UserProxyAgent(
        name="User",
        human_input_mode="NEVER",
        max_consecutive_auto_reply=1,
        code_execution_config={"work_dir": "coding", "use_docker": False}
    )

    # 2. Connect and Register Tools using the 'mcps' config
    # This keeps the connections open for the duration of the workflow
    active_connections = await connect_mcp_tools(ebs_admin, user_proxy, mcps)

    try:
        # 3. Start the Conversation
        await user_proxy.a_initiate_chat(
            ebs_admin,
            message="Submit a request for program 'OEXOEORD' in application 'ONT' and check its status."
        )
    finally:
        # Cleanup connections
        for client, session in active_connections:
            await session.__aexit__(None, None, None)
            await client.__aexit__(None, None, None)

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

📋 How it Works

Configuration: The mcps list defines where the agents look for tools.
Connection: The connect_mcp_tools function iterates over mcps, performs the SSE handshake, and keeps the session alive.
Registration: Tools discovered via session.list_tools() are dynamically wrapped and registered using register_for_llm (for the Assistant) and register_for_execution (for the Proxy).
Execution: When the OracleAdmin agent calls submit_ebs_request, the wrapper routes the call via the active MCP session to the Docker container.

🛡️ Quality Assurance

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

Transparency: This page may contain affiliate links.