Welcome to the final part of my 3-part series on integrating Oracle APEX with AI using Model Context Protocol (MCP).
In Part I, I introduced the architecture and foundations. In Part II, I walked you through building and connecting the MCP server.
Now in Part III, I’ll show you how to bring everything together with advanced integrations, security considerations, and deployment patterns — so you can take your APEX + AI projects from prototype to production.
The MCP Gateway acts as a bridge between Claude Desktop and your MCP server. Here's how to configure it:
# Create catalogs directory
mkdir -p ~/.docker/mcp/catalogs
# Create custom.yaml
nano ~/.docker/mcp/catalogs/custom.yaml
Add your server configuration:
version: 2
name: custom
displayName: Custom MCP Servers
registry:
apex-hr:
description: "Oracle APEX HR Employee Management with REST API integration"
title: "APEX HR Management"
type: server
dateAdded: "2024-12-16T00:00:00Z"
image: apex-hr-mcp-server:latest
ref: ""
readme: ""
toolsUrl: ""
source: ""
upstream: ""
icon: ""
tools:
- name: get_employee_info
- name: get_employee_details
- name: get_employee_by_id
- name: get_employee_by_name
- name: create_employee
metadata:
category: integration
tags:
- oracle
- apex
- hr
- enterprise
- database
license: MIT
owner: local
# Edit registry file
nano ~/.docker/mcp/registry.yamlAdd under the registry: key:
registry:
# ... existing servers ...
apex-hr:
ref: ""Find your Claude Desktop config:
~/Library/Application Support/Claude/claude_desktop_config.json%APPDATA%\Claude\claude_desktop_config.json~/.config/Claude/claude_desktop_config.json
{
"mcpServers": {
"mcp-toolkit-gateway": {
"command": "docker",
"args": [
"run",
"-i",
"--rm",
"-v", "/var/run/docker.sock:/var/run/docker.sock",
"-v", "/Users/your_username/.docker/mcp:/mcp",
"docker/mcp-gateway",
"--catalog=/mcp/catalogs/docker-mcp.yaml",
"--catalog=/mcp/catalogs/custom.yaml",
"--config=/mcp/config.yaml",
"--registry=/mcp/registry.yaml",
"--tools-config=/mcp/tools.yaml",
"--transport=stdio"
]
}
}
}
Important: Replace /Users/your_username with your actual home directory path.
Try these natural language queries:
"What HR tools do you have available?"
"Show me all employee information"
"Find employee with ID 7369"
"Look up employee named SMITH"
"Create a new employee named Alice Johnson as a MANAGER with salary 5000 in department 10"Replace the base URL and endpoints:
# Replace this
BASE_URL = "https://your-apex-instance.com/apexea/demo/hr"
# With your APEX instance
BASE_URL = "https://your-apex-instance.com/ords/workspace/api"For production APEX apps, add authentication:
# Add to your server configuration
APEX_USERNAME = os.environ.get("APEX_USERNAME", "")
APEX_PASSWORD = os.environ.get("APEX_PASSWORD", "")
# In your HTTP calls
auth = httpx.BasicAuth(APEX_USERNAME, APEX_PASSWORD)
response = await client.get(url, auth=auth, timeout=10)
# Set up authentication
docker mcp secret set APEX_USERNAME="your-username"
docker mcp secret set APEX_PASSWORD="your-password"Update your catalog to include secrets:
secrets:
- name: APEX_USERNAME
env: APEX_USERNAME
example: "demo_user"
- name: APEX_PASSWORD
env: APEX_PASSWORD
example: "demo_password"For APEX applications with complex data structures:
def format_complex_data(data):
"""Handle nested objects and arrays from APEX"""
if isinstance(data, dict):
if 'items' in data: # APEX REST list response
return format_apex_list(data['items'])
else:
return format_apex_object(data)
return str(data)
def format_apex_list(items):
"""Format APEX REST list responses"""
if not items:
return "No items found"
formatted = []
for i, item in enumerate(items, 1):
formatted.append(f"\n📋 Item #{i}:")
for key, value in item.items():
# Handle APEX naming conventions
display_key = key.replace('_', ' ').title()
formatted.append(f" - {display_key}: {value}")
return "\n".join(formatted)For bulk data operations:
@mcp.tool()
async def bulk_employee_update(employee_data: str = "") -> str:
"""Update multiple employees from CSV-like data."""
try:
# Parse the input (could be CSV, JSON, etc.)
updates = parse_bulk_data(employee_data)
results = []
async with httpx.AsyncClient() as client:
for update in updates:
try:
response = await client.put(
f"{BASE_URL}/employees/{update['id']}",
json=update['data'],
timeout=10
)
response.raise_for_status()
results.append(f"✅ Updated employee {update['id']}")
except Exception as e:
results.append(f"❌ Failed to update {update['id']}: {str(e)}")
return "\n".join(results)
except Exception as e:
return f"❌ Bulk update failed: {str(e)}"
# Use environment variables for credentials
APEX_CLIENT_ID = os.environ.get("APEX_CLIENT_ID", "")
APEX_CLIENT_SECRET = os.environ.get("APEX_CLIENT_SECRET", "")
# Implement OAuth2 if available
async def get_oauth_token():
"""Get OAuth2 token for APEX REST APIs"""
token_url = f"{BASE_URL}/oauth/token"
data = {
"grant_type": "client_credentials",
"client_id": APEX_CLIENT_ID,
"client_secret": APEX_CLIENT_SECRET
}
async with httpx.AsyncClient() as client:
response = await client.post(token_url, data=data)
response.raise_for_status()
return response.json()["access_token"]
import re
def sanitize_input(input_str):
"""Sanitize user input to prevent injection attacks"""
# Remove potentially dangerous characters
sanitized = re.sub(r'[<>"\';]', '', input_str)
# Limit length
return sanitized[:100]
@mcp.tool()
async def safe_employee_search(name: str = "") -> str:
"""Search with input sanitization"""
clean_name = sanitize_input(name)
# ... rest of function
import asyncio
from collections import defaultdict
from datetime import datetime, timedelta
class RateLimiter:
def __init__(self, max_requests=100, window_minutes=60):
self.max_requests = max_requests
self.window = timedelta(minutes=window_minutes)
self.requests = defaultdict(list)
async def check_rate_limit(self, user_id="default"):
now = datetime.now()
# Clean old requests
self.requests[user_id] = [
req_time for req_time in self.requests[user_id]
if now - req_time < self.window
]
if len(self.requests[user_id]) >= self.max_requests:
return False
self.requests[user_id].append(now)
return True
# Use in your tools
rate_limiter = RateLimiter()
@mcp.tool()
async def rate_limited_tool() -> str:
"""Tool with rate limiting"""
if not await rate_limiter.check_rate_limit():
return "❌ Rate limit exceeded. Please try again later."
# ... rest of function
# config.py
import os
class Config:
def __init__(self):
self.environment = os.environ.get("ENVIRONMENT", "development")
self.base_url = self._get_base_url()
self.timeout = int(os.environ.get("API_TIMEOUT", "10"))
self.max_retries = int(os.environ.get("MAX_RETRIES", "3"))
def _get_base_url(self):
urls = {
"development": "https://dev-apex.company.com/ords/hr/api",
"staging": "https://staging-apex.company.com/ords/hr/api",
"production": "https://apex.company.com/ords/hr/api"
}
return urls.get(self.environment, urls["development"])
config = Config()
import structlog
# Enhanced logging
logger = structlog.get_logger()
@mcp.tool()
async def monitored_tool(param: str = "") -> str:
"""Tool with comprehensive monitoring"""
start_time = datetime.now()
try:
logger.info(
"tool_execution_started",
tool="monitored_tool",
param=param,
timestamp=start_time.isoformat()
)
# ... tool implementation
end_time = datetime.now()
duration = (end_time - start_time).total_seconds()
logger.info(
"tool_execution_completed",
tool="monitored_tool",
duration_seconds=duration,
status="success"
)
return result
except Exception as e:
logger.error(
"tool_execution_failed",
tool="monitored_tool",
error=str(e),
duration_seconds=(datetime.now() - start_time).total_seconds()
)
raise
@mcp.tool()
async def health_check() -> str:
"""Check the health of APEX API connections"""
checks = []
# Test database connectivity
try:
async with httpx.AsyncClient() as client:
response = await client.get(f"{BASE_URL}/health", timeout=5)
response.raise_for_status()
checks.append("✅ APEX API: Connected")
except Exception as e:
checks.append(f"❌ APEX API: {str(e)}")
# Test authentication
try:
token = await get_oauth_token()
checks.append("✅ Authentication: Valid")
except Exception as e:
checks.append(f"❌ Authentication: {str(e)}")
return "🔍 Health Check Results:\n" + "\n".join(checks)Symptoms: Claude doesn't recognize your tools
Solutions:
# Check if Docker image built successfully
docker images | grep apex-hr-mcp-server
# Verify catalog configuration
cat ~/.docker/mcp/catalogs/custom.yaml
# Check Claude Desktop config
cat "~/Library/Application Support/Claude/claude_desktop_config.json"
# Restart Claude Desktop completelySymptoms: HTTP errors when calling APEX APIs
Debugging:
# Add detailed error logging
except httpx.HTTPStatusError as e:
logger.error(f"HTTP {e.response.status_code}: {e.response.text}")
logger.error(f"Request URL: {e.request.url}")
logger.error(f"Request headers: {e.request.headers}")
return f"❌ API Error: {e.response.status_code}"Symptoms: 401/403 errors from APEX
Solutions:
# Verify secrets are set
docker mcp secret list
# Test credentials directly
curl -u "username:password" "https://your-apex.com/ords/api/test"
# Check APEX workspace permissionsSymptoms: Slow response times
Optimizations:
# Use connection pooling
async with httpx.AsyncClient(
limits=httpx.Limits(max_connections=10, max_keepalive_connections=5)
) as client:
# ... requests
# Implement caching
from functools import lru_cache
@lru_cache(maxsize=100)
def get_cached_employee(employee_id):
# Cache frequently accessed data
pass
# Add timeouts
response = await client.get(url, timeout=httpx.Timeout(5.0))
@mcp.tool()
async def create_sales_opportunity(
customer_name: str = "",
amount: str = "",
close_date: str = "",
probability: str = ""
) -> str:
"""Create a new sales opportunity in the CRM system"""
# Validation
if not customer_name.strip():
return "❌ Customer name is required"
try:
amount_float = float(amount) if amount else 0
prob_int = int(probability) if probability else 50
except ValueError:
return "❌ Amount must be numeric, probability must be integer 0-100"
payload = {
"customer_name": customer_name,
"opportunity_amount": amount_float,
"close_date": close_date,
"win_probability": prob_int,
"stage": "Prospecting",
"created_by": "AI_Assistant"
}
try:
async with httpx.AsyncClient() as client:
response = await client.post(
f"{BASE_URL}/opportunities",
json=payload,
timeout=10
)
response.raise_for_status()
data = response.json()
opp_id = data.get('opportunity_id', 'Unknown')
return f"✅ Sales opportunity created! ID: {opp_id}\n💰 Amount: ${amount_float:,.2f}\n📅 Target Close: {close_date}"
except Exception as e:
return f"❌ Failed to create opportunity: {str(e)}"
@mcp.tool()
async def get_project_status(project_id: str = "") -> str:
"""Get comprehensive project status including tasks, timeline, and budget"""
if not project_id.strip():
return "❌ Project ID is required"
try:
async with httpx.AsyncClient() as client:
# Get project details
project_response = await client.get(f"{BASE_URL}/projects/{project_id}")
project_response.raise_for_status()
project = project_response.json()
# Get tasks
tasks_response = await client.get(f"{BASE_URL}/projects/{project_id}/tasks")
tasks_response.raise_for_status()
tasks = tasks_response.json()
# Get budget info
budget_response = await client.get(f"{BASE_URL}/projects/{project_id}/budget")
budget_response.raise_for_status()
budget = budget_response.json()
# Calculate metrics
total_tasks = len(tasks.get('items', []))
completed_tasks = len([t for t in tasks.get('items', []) if t.get('status') == 'Completed'])
completion_rate = (completed_tasks / total_tasks * 100) if total_tasks > 0 else 0
return f"""📊 Project Status Report - {project.get('name', 'Unknown')}
🎯 **Progress**: {completion_rate:.1f}% ({completed_tasks}/{total_tasks} tasks)
📅 **Timeline**: {project.get('start_date')} → {project.get('end_date')}
💰 **Budget**: ${budget.get('spent', 0):,.2f} / ${budget.get('total', 0):,.2f}
👥 **Team Size**: {project.get('team_size', 0)} members
⚠️ **Status**: {project.get('status', 'Unknown')}
🔥 **Critical Tasks**:
{get_critical_tasks(tasks.get('items', []))}"""
except Exception as e:
return f"❌ Error retrieving project status: {str(e)}"
def get_critical_tasks(tasks):
"""Extract and format critical/overdue tasks"""
critical = [t for t in tasks if t.get('priority') == 'High' or t.get('overdue', False)]
if not critical:
return "✅ No critical tasks"
formatted = []
for task in critical[:5]: # Limit to top 5
status = "⚠️ OVERDUE" if task.get('overdue') else "🔥 HIGH PRIORITY"
formatted.append(f"- {status}: {task.get('name', 'Unnamed task')}")
return "\n".join(formatted)
@mcp.tool()
async def generate_financial_summary(period: str = "current_month") -> str:
"""Generate financial summary report for the specified period"""
period_map = {
"current_month": "CM",
"last_month": "LM",
"current_quarter": "CQ",
"last_quarter": "LQ",
"current_year": "CY"
}
if period not in period_map:
return f"❌ Invalid period. Use: {', '.join(period_map.keys())}"
try:
async with httpx.AsyncClient() as client:
# Get financial summary
response = await client.get(
f"{BASE_URL}/financials/summary",
params={"period": period_map[period]}
)
response.raise_for_status()
data = response.json()
revenue = data.get('revenue', 0)
expenses = data.get('expenses', 0)
net_income = revenue - expenses
margin = (net_income / revenue * 100) if revenue > 0 else 0
# Get comparison data
comparison = data.get('comparison', {})
revenue_change = comparison.get('revenue_change_pct', 0)
trend_indicator = "📈" if revenue_change > 0 else "📉" if revenue_change < 0 else "➡️"
return f"""💰 Financial Summary - {period.replace('_', ' ').title()}
📊 **Revenue**: ${revenue:,.2f}
💸 **Expenses**: ${expenses:,.2f}
💎 **Net Income**: ${net_income:,.2f}
📈 **Margin**: {margin:.1f}%
{trend_indicator} **vs Previous Period**: {revenue_change:+.1f}%
🏆 **Top Revenue Sources**:
{format_revenue_sources(data.get('revenue_sources', []))}
⚠️ **Expense Breakdown**:
{format_expense_breakdown(data.get('expense_breakdown', []))}"""
except Exception as e:
return f"❌ Error generating financial summary: {str(e)}"Combine multiple APEX applications:
# Multi-app configuration
APPS = {
"hr": "https://apex.company.com/ords/hr/api",
"crm": "https://apex.company.com/ords/crm/api",
"finance": "https://apex.company.com/ords/finance/api"
}
@mcp.tool()
async def cross_app_employee_report(employee_id: str = "") -> str:
"""Get comprehensive employee report across HR, CRM, and Finance systems"""
results = {}
# Get HR data
results['hr'] = await get_hr_data(employee_id)
# Get CRM data (sales performance)
results['crm'] = await get_crm_data(employee_id)
# Get Finance data (compensation, expenses)
results['finance'] = await get_finance_data(employee_id)
return format_cross_app_report(results)
@mcp.tool()
async def automate_new_hire_process(
employee_name: str = "",
department: str = "",
position: str = "",
start_date: str = ""
) -> str:
"""Automate the complete new hire process across multiple systems"""
steps = []
try:
# Step 1: Create employee record
emp_result = await create_employee_record(employee_name, department, position)
steps.append(f"✅ Employee record created: {emp_result}")
# Step 2: Setup system accounts
account_result = await setup_system_accounts(employee_name)
steps.append(f"✅ System accounts: {account_result}")
# Step 3: Assign equipment
equipment_result = await assign_equipment(employee_name, department)
steps.append(f"✅ Equipment assigned: {equipment_result}")
# Step 4: Create onboarding tasks
tasks_result = await create_onboarding_tasks(employee_name, start_date)
steps.append(f"✅ Onboarding tasks: {tasks_result}")
# Step 5: Send notifications
notify_result = await send_notifications(employee_name, department)
steps.append(f"✅ Notifications sent: {notify_result}")
return f"🎉 New hire process completed for {employee_name}!\n\n" + "\n".join(steps)
except Exception as e:
return f"❌ New hire process failed at step {len(steps) + 1}: {str(e)}\n\nCompleted steps:\n" + "\n".join(steps)
@mcp.tool()
async def ai_performance_insights(department: str = "") -> str:
"""Generate AI-powered insights about department performance"""
try:
# Gather data from multiple sources
performance_data = await get_performance_metrics(department)
financial_data = await get_financial_metrics(department)
employee_data = await get_employee_metrics(department)
# Use AI to generate insights
insights = await generate_ai_insights({
'performance': performance_data,
'financial': financial_data,
'employee': employee_data
})
return format_insights_report(insights, department)
except Exception as e:
return f"❌ Error generating insights: {str(e)}"
async def generate_ai_insights(data):
"""Use AI to analyze data and generate insights"""
# This could integrate with OpenAI, local AI models, etc.
# For now, implement rule-based insights
insights = []
# Performance insights
if data['performance']['efficiency'] < 80:
insights.append({
'type': 'warning',
'message': 'Department efficiency below target (80%)',
'recommendation': 'Consider process automation or additional training'
})
# Financial insights
if data['financial']['budget_variance'] > 10:
insights.append({
'type': 'alert',
'message': f"Budget variance of {data['financial']['budget_variance']}%",
'recommendation': 'Review spending patterns and adjust budget allocation'
})
return insightsYou've now built a complete enterprise-grade integration between Oracle APEX and modern LLMs using Model Context Protocol. This isn't just a technical exercise—it's the foundation for transforming how your organization interacts with its data and systems.
This integration represents a fundamental shift in enterprise software interaction. Instead of training users on complex interfaces, you're bringing AI to your existing systems. Your employees can now:
The future of enterprise software is conversational, intelligent, and agent-driven. You now have the tools and knowledge to make that future a reality in your organization.
Ready to transform your enterprise with AI? Start building your first MCP server today.
Want to dive deeper? Check out the Oracle APEX REST API documentation and the Model Context Protocol specification for advanced patterns and use cases.
This comprehensive tutorial walks you through creating production-ready integrations between Oracle APEX applications and Large Language Models using the Model Context Protocol (MCP). The patterns and code examples are battle-tested and ready for enterprise deployment.
Last Updated: September 2025
That wraps up my 3-part series on Oracle APEX and AI integration with MCP. We’ve covered the foundations, the server setup, and now advanced deployments — giving you a full roadmap to build intelligent, agentic applications.
I hope this series has inspired you to experiment with APEX + AI and see how MCP can unlock new possibilities in your projects. Thanks for following along!