How to Create Intelligent, Conversational Interfaces for Your Enterprise Oracle APEX Applications Using Model Context Protocol
In Part I of this series, we explored the power of connecting Oracle APEX applications to Large Language Models (LLMs) using the Model Context Protocol (MCP) — and why this integration is transforming the way enterprises build intelligent, conversational interfaces.
Now, it’s time to roll up our sleeves and get hands-on. In Part II, we’ll walk step-by-step through the process of building your MCP server — the crucial bridge that allows APEX to communicate with your AI. You’ll learn how to connect REST APIs, structure your project, set up Docker, and write server logic that turns natural-language requests into real-world actions.
Prerequisites
Before we start, make sure you have:
- Docker Desktop installed with MCP Toolkit enabled
- Claude Desktop (or another MCP-compatible LLM interface)
- Oracle APEX application with REST APIs enabled
- Basic Python knowledge (I'll explain everything step-by-step)
Step 1: Understanding Your APEX REST APIs
First, you need to identify and document your APEX REST APIs. For our example, we're using these endpoints:
# Base URL
https://your-apex-instance.com/apexea/demo/hr
# Available endpoints:
GET /empinfo/ # General employee information
GET /employees/ # Detailed employee data
GET /employees/:id # Employee by ID
GET /empsec/:empname # Employee by name
POST /emp_post_example/ # Create new employee
Finding Your APEX REST APIs
In your APEX workspace:
- Go to SQL Workshop → RESTful Services
- Review your Resource Templates
- Note the HTTP Methods and Parameters
- Test endpoints using the built-in testing tools
- Document the Request/Response formats
Step 2: Project Structure Setup
Create your project directory and organize the files:
# Create project directory
mkdir apex-hr-mcp-server
cd apex-hr-mcp-server
# We'll create these files:
# ├── Dockerfile
# ├── requirements.txt
# ├── hr_employee_server.py
# ├── readme.txt
# └── CLAUDE.md
Step 3: Creating the Docker Environment
Dockerfile
Create a Dockerfile that sets up our Python environment:
# Use Python slim image for efficiency
FROM python:3.11-slim
# Set working directory
WORKDIR /app
# Set Python unbuffered mode for better logging
ENV PYTHONUNBUFFERED=1
# Copy requirements first for better Docker layer caching
COPY requirements.txt .
# Install Python dependencies
RUN pip install --no-cache-dir -r requirements.txt
# Copy the server code
COPY hr_employee_server.py .
# Create non-root user for security
RUN useradd -m -u 1000 mcpuser && \
chown -R mcpuser:mcpuser /app
# Switch to non-root user
USER mcpuser
# Run the server
CMD ["python", "hr_employee_server.py"]
Requirements
Create requirements.txt with our dependencies:
mcp[cli]>=1.2.0
httpxThe mcp package provides the FastMCP framework, and httpx handles our HTTP requests to APEX APIs.
Step 4: Building the MCP Server
Now for the main event—our MCP server code. Create hr_employee_server.py:
#!/usr/bin/env python3
"""
HR Employee Management MCP Server - Provides access to Oracle APEX HR APIs
"""
import os
import sys
import logging
from datetime import datetime, timezone
import httpx
from mcp.server.fastmcp import FastMCP
# Configure logging to stderr for Docker
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
stream=sys.stderr
)
logger = logging.getLogger("hr-employee-server")
# Initialize MCP server
mcp = FastMCP("hr-employee")
# Configuration - replace with your APEX base URL
BASE_URL = "https://your-apex-instance.com/apexea/demo/hr"
# === UTILITY FUNCTIONS ===
def format_employee_data(data):
"""Format employee data for readable display"""
if isinstance(data, dict):
formatted = []
for key, value in data.items():
formatted.append(f"- {key}: {value}")
return "\n".join(formatted)
elif isinstance(data, list):
if not data:
return "No employees found"
formatted = []
for i, item in enumerate(data, 1):
if isinstance(item, dict):
formatted.append(f"\n👤 Employee #{i}:")
for key, value in item.items():
formatted.append(f" - {key}: {value}")
else:
formatted.append(f" - {item}")
return "\n".join(formatted)
return str(data)
# === MCP TOOLS ===
@mcp.tool()
async def get_employee_info() -> str:
"""Get general information about all employees in the system."""
logger.info("Fetching general employee information")
try:
async with httpx.AsyncClient() as client:
response = await client.get(f"{BASE_URL}/empinfo/", timeout=10)
response.raise_for_status()
data = response.json()
formatted_data = format_employee_data(data)
return f"✅ Employee Information:\n{formatted_data}"
except httpx.HTTPStatusError as e:
logger.error(f"HTTP error: {e.response.status_code}")
return f"❌ API Error: {e.response.status_code} - {e.response.text}"
except Exception as e:
logger.error(f"Error fetching employee info: {e}")
return f"❌ Error: {str(e)}"
@mcp.tool()
async def get_employee_details() -> str:
"""Get detailed information about employees including salaries, jobs, managers, and departments."""
logger.info("Fetching detailed employee information")
try:
async with httpx.AsyncClient() as client:
response = await client.get(f"{BASE_URL}/employees/", timeout=10)
response.raise_for_status()
data = response.json()
formatted_data = format_employee_data(data)
return f"💼 Employee Details:\n{formatted_data}"
except httpx.HTTPStatusError as e:
logger.error(f"HTTP error: {e.response.status_code}")
return f"❌ API Error: {e.response.status_code} - {e.response.text}"
except Exception as e:
logger.error(f"Error fetching employee details: {e}")
return f"❌ Error: {str(e)}"
@mcp.tool()
async def get_employee_by_id(employee_id: str = "") -> str:
"""Get information about a specific employee using their ID number."""
logger.info(f"Fetching employee by ID: {employee_id}")
if not employee_id.strip():
return "❌ Error: Employee ID is required"
try:
async with httpx.AsyncClient() as client:
response = await client.get(f"{BASE_URL}/employees/{employee_id}", timeout=10)
response.raise_for_status()
data = response.json()
formatted_data = format_employee_data(data)
return f"👤 Employee ID {employee_id}:\n{formatted_data}"
except httpx.HTTPStatusError as e:
logger.error(f"HTTP error: {e.response.status_code}")
if e.response.status_code == 404:
return f"❌ Employee with ID {employee_id} not found"
return f"❌ API Error: {e.response.status_code} - {e.response.text}"
except Exception as e:
logger.error(f"Error fetching employee by ID: {e}")
return f"❌ Error: {str(e)}"
@mcp.tool()
async def get_employee_by_name(employee_name: str = "") -> str:
"""Get information about a specific employee using their name."""
logger.info(f"Fetching employee by name: {employee_name}")
if not employee_name.strip():
return "❌ Error: Employee name is required"
try:
async with httpx.AsyncClient() as client:
response = await client.get(f"{BASE_URL}/empsec/{employee_name}", timeout=10)
response.raise_for_status()
data = response.json()
formatted_data = format_employee_data(data)
return f"👤 Employee '{employee_name}':\n{formatted_data}"
except httpx.HTTPStatusError as e:
logger.error(f"HTTP error: {e.response.status_code}")
if e.response.status_code == 404:
return f"❌ Employee with name '{employee_name}' not found"
return f"❌ API Error: {e.response.status_code} - {e.response.text}"
except Exception as e:
logger.error(f"Error fetching employee by name: {e}")
return f"❌ Error: {str(e)}"
@mcp.tool()
async def create_employee(name: str = "", job: str = "", manager: str = "", salary: str = "", commission: str = "", department: str = "") -> str:
"""Create a new employee with the specified details (name, job, manager ID, salary, commission, department number)."""
logger.info(f"Creating new employee: {name}")
# Input validation
if not name.strip():
return "❌ Error: Employee name is required"
if not job.strip():
return "❌ Error: Job title is required"
# Validate numeric fields
if salary.strip():
try:
float(salary)
except ValueError:
return "❌ Error: Salary must be a number"
if manager.strip():
try:
int(manager)
except ValueError:
return "❌ Error: Manager must be a valid employee ID (number)"
if department.strip():
try:
int(department)
except ValueError:
return "❌ Error: Department must be a valid department number"
try:
# Prepare the payload for APEX
payload = {
"ename": name.strip(),
"ejob": job.strip(),
"mgr": manager.strip() if manager.strip() else None,
"sal": salary.strip() if salary.strip() else None,
"comm": commission.strip() if commission.strip() else None,
"deptno": department.strip() if department.strip() else None
}
async with httpx.AsyncClient() as client:
response = await client.post(
f"{BASE_URL}/emp_post_example/",
json=payload,
timeout=10
)
# Handle APEX response codes
if response.status_code == 201:
try:
result_data = response.json()
if "forward_location" in result_data:
new_employee_id = result_data["forward_location"].split("/")[-1]
return f"✅ Employee created successfully! New employee ID: {new_employee_id}"
else:
return f"✅ Employee '{name}' created successfully!"
except:
return f"✅ Employee '{name}' created successfully!"
elif response.status_code == 400:
try:
error_data = response.json()
error_msg = error_data.get("errmsg", "Bad request")
return f"❌ Creation failed: {error_msg}"
except:
return f"❌ Creation failed: Invalid request data"
else:
response.raise_for_status()
except httpx.HTTPStatusError as e:
logger.error(f"HTTP error: {e.response.status_code}")
try:
error_data = e.response.json()
error_msg = error_data.get("errmsg", e.response.text)
return f"❌ API Error: {error_msg}"
except:
return f"❌ API Error: {e.response.status_code} - {e.response.text}"
except Exception as e:
logger.error(f"Error creating employee: {e}")
return f"❌ Error: {str(e)}"
# === SERVER STARTUP ===
if __name__ == "__main__":
logger.info("Starting HR Employee Management MCP server...")
try:
mcp.run(transport='stdio')
except Exception as e:
logger.error(f"Server error: {e}", exc_info=True)
sys.exit(1)
Step 5: Understanding the Code Structure
Let's break down the key components:
1. MCP Tool Decorators
Each function decorated with @mcp.tool() becomes available to the LLM:
@mcp.tool()
async def get_employee_by_id(employee_id: str = "") -> str:
"""Get information about a specific employee using their ID number."""Key Requirements:
- Single-line docstrings only (multi-line breaks Claude Desktop)
- Parameters must default to empty strings, not
None - Must return strings (for proper display)
- Async functions for better performance
2. Error Handling Pattern
Every tool follows this pattern:
try:
# API call
async with httpx.AsyncClient() as client:
response = await client.get(url, timeout=10)
response.raise_for_status()
data = response.json()
# Format and return data
return f"✅ Success: {formatted_data}"
except httpx.HTTPStatusError as e:
# Handle HTTP errors
return f"❌ API Error: {e.response.status_code}"
except Exception as e:
# Handle all other errors
return f"❌ Error: {str(e)}"3. Input Validation
Critical for enterprise applications:
# Check required fields
if not employee_id.strip():
return "❌ Error: Employee ID is required"
# Validate numeric fields
try:
float(salary)
except ValueError:
return "❌ Error: Salary must be a number"Step 6: Docker Configuration Files
Building the Docker Image
# Build the image
docker build -t apex-hr-mcp-server .
# Verify the build
docker images | grep apex-hr-mcp-server
By now, you have a fully functional MCP server containerized and ready to interact with your APEX application — the critical foundation for any agentic AI workflow. But this is just the beginning.
In Part III, we’ll take things to the next level by integrating your MCP server with tools like Claude Desktop and implementing enterprise-grade enhancements like authentication, security, and scalability — so you can confidently deploy agentic applications in production.
Ready to Take Your Oracle Skills to the Next Level?
Master Oracle APEX – Join our Hands-On Intro to Oracle APEX course on OraPub and build powerful applications with expert guidance.
Check out Viscosity’s event page for upcoming virtual and on-site training opportunities.
SUBMIT YOUR COMMENT