OCI AI Solutions Engineer Interview Question And Answer part2

 Project Overview: Autonomous Database MCP Server

The Autonomous AI Database MCP Server acts as a stateless, highly secure, and standardized JSON-RPC 2.0 bridge. It exposes structured enterprise database capabilities to any MCP-compliant LLM client (such as Claude Desktop, VS Code Cline, or OCI AI Agents) without rewriting integration logic. Instead of giving the AI raw, risky SQL execution privileges, it forces the AI to use strictly typed tools governed by the database's inner RBAC, schema limits, and Oracle DBMS_CLOUD_AI_AGENT (Select AI) framework.

🛠️ Tech Stack & Dependencies

Language: Python 3.11+
Protocol Core: mcp (Official Python Model Context Protocol SDK)
Database Connectivity: oracledb (Thin Driver for Oracle Autonomous Database / 26ai)
Asynchronous Runtime: asyncio & pydantic (for type validations)

 Project Directory Structure

text

autonomous-db-mcp/
├── server.py              # Primary MCP Server code implementing tools
├── test_mcp_server.py     # Component and integration test cases
├── requirements.txt       # Project dependencies
└── README.md              # Setup and execution guide

 Core Project Code (server.py)

This standalone implementation creates an asynchronous MCP server that safely exposes database schema exploration and contextual queries via standard input/output (stdio).

python

import asyncio
import os
from mcp.server.models import InitializationOptions
import mcp.types as types
from mcp.server import NotificationOptions, Server
import mcp.server.stdio
import oracledb

# Initialize the MCP Server
server = Server("autonomous-db-mcp-server")

# DB Connection Helper using environment variables
def get_db_connection():
    return oracledb.connect(
        user=os.getenv("DB_USER", "ADMIN"),
        password=os.getenv("DB_PASSWORD"),
        dsn=os.getenv("DB_DSN"),  # e.g., "myautonomousdb_high"
        config_dir=os.getenv("WALLET_DIR", None)
    )

@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
    """Advertise available database tools to the AI Client."""
    return [
        types.Tool(
            name="get_schema_insight",
            description="Retrieves a list of accessible tables and column descriptions to help understand the database structure.",
            inputSchema={
                "type": "object",
                "properties": {
                    "schema_name": {"type": "string", "description": "The target DB schema name (uppercase)."}
                },
                "required": ["schema_name"]
            }
        ),
        types.Tool(
            name="query_autonomous_agent",
            description="Executes a safe, natural language query against the database using the internal Select AI agent.",
            inputSchema={
                "type": "object",
                "properties": {
                    "natural_query": {"type": "string", "description": "The plain text question regarding database data."}
                },
                "required": ["natural_query"]
            }
        )
    ]

@server.call_tool()
async def handle_call_tool(name: str, arguments: dict | None) -> list[types.TextContent]:
    """Execute the concrete database logic requested by the AI client."""
    if not arguments:
        raise ValueError("Missing arguments for tool execution")

    try:
        connection = get_db_connection()
        cursor = connection.cursor()

        if name == "get_schema_insight":
            schema = arguments.get("schema_name", "ADMIN").upper()
            # Constrained metadata query preventing arbitrary SQL execution
            cursor.execute(
                "SELECT table_name, column_name, data_type FROM all_tab_columns WHERE owner = :schema AND rownum <= 50",
                schema=schema
            )
            rows = cursor.fetchall()
            result_str = "\n".join([f"Table: {r[0]}, Column: {r[1]}, Type: {r[2]}" for r in rows])
            
            cursor.close()
            connection.close()
            return [types.TextContent(type="text", text=result_str or "No matching schema items found.")]

        elif name == "query_autonomous_agent":
            query_str = arguments.get("natural_query")
            # Leverages Oracle Autonomous DB DBMS_CLOUD_AI_AGENT for secure NL-to-SQL translation
            cursor.execute(
                "SELECT DBMS_CLOUD_AI_AGENT.RUN_QUERY('MY_DB_AGENT', :query) FROM dual",
                query=query_str
            )
            result = cursor.fetchone()[0]
            
            cursor.close()
            connection.close()
            return [types.TextContent(type="text", text=str(result))]
            
        else:
            raise ValueError(f"Unknown tool requested: {name}")

    except Exception as e:
        return [types.TextContent(type="text", text=f"Execution Error: {str(e)}")]

async def main():
    # Run the server using local standard input/output transport
    async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            InitializationOptions(
                server_name="autonomous-db-mcp-server",
                server_version="1.0.0",
                capabilities=server.get_capabilities(
                    notification_options=NotificationOptions(),
                    experimental_capabilities={},
                ),
            ),
        )

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

 Commands: Setup, Execution, and Deployment

Run these tasks sequentially in your terminal to initialize and run the server environment:

bash

# 1. Create a Python virtual environment and activate it
python -m venv venv
source venv/bin/activate  # On Windows use: venv\Scripts\activate

# 2. Install required packages
pip install mcp oracledb pytest pytest-asyncio

# 3. Export environment variables for Autonomous DB Authentication
export DB_USER="ADMIN"
export DB_PASSWORD="YourSecurePassword123!"
export DB_DSN="yourdbname_high"
export WALLET_DIR="/path/to/secure/wallet"

# 4. Start the MCP Server locally over stdio transport
python server.py

# 5. Optional: Run tests using pytest
pytest test_mcp_server.py -v

 Automated Test Cases (test_mcp_server.py)

This test suite utilizes mock objects to isolate database connectivity and cleanly verify tool registration and validation logic.

python

import pytest
from unittest.mock import patch, MagicMock
import mcp.types as types
from server import handle_list_tools, handle_call_tool

@pytest.mark.asyncio
async def test_handle_list_tools():
    """Verify the server correctly advertises database capabilities."""
    tools = await handle_list_tools()
    assert len(tools) == 2
    assert tools[0].name == "get_schema_insight"
    assert tools[1].name == "query_autonomous_agent"

@pytest.mark.asyncio
@patch('server.get_db_connection')
async def test_handle_call_tool_schema_insight(mock_get_db):
    """Verify metadata processing logic for get_schema_insight."""
    # Setup mock connections
    mock_conn = MagicMock()
    mock_cursor = MagicMock()
    mock_get_db.return_value = mock_conn
    mock_conn.cursor.return_value = mock_cursor
    mock_cursor.fetchall.return_value = [("EMPLOYEES", "EMP_ID", "NUMBER")]

    arguments = {"schema_name": "HR"}
    response = await handle_call_tool("get_schema_insight", arguments)

    assert len(response) == 1
    assert isinstance(response[0], types.TextContent)
    assert "Table: EMPLOYEES, Column: EMP_ID, Type: NUMBER" in response[0].text
    mock_cursor.execute.assert_called_once()

@pytest.mark.asyncio
@patch('server.get_db_connection')
async def test_handle_call_tool_execution_error(mock_get_db):
    """Verify the server captures exceptions gracefully and formats them as standard tool text outputs."""
    mock_get_db.side_effect = Exception("Database connection timeout")

    arguments = {"natural_query": "Show quarterly sales summary"}
    response = await handle_call_tool("query_autonomous_agent", arguments)

    assert len(response) == 1
    assert "Execution Error: Database connection timeout" in response[0].text

 Interview Questions & Answers

Q1: What is the core problem that the Model Context Protocol (MCP) addresses when connecting LLMs to an Autonomous Database?

Answer: Without MCP, linking an AI application to a database creates an M × N integration bottleneck, where every custom application template requires unique API wrapper functions and connector code for every targeted database engine.

MCP shifts this architectural paradigm to a standardized M + N model. The database exposes its capabilities natively through a unified MCP server endpoint. Any MCP-compliant client can instantly query schemas or request actions using standardized JSON-RPC 2.0 communication protocols, eliminating custom glue code entirely.

Q2: How should you handle security and guardrails when exposing a live enterprise database to an AI Agent via MCP?

Answer: To safely implement AI-to-database communication:

Never allow arbitrary SQL string execution: Prevent the client from submitting raw SQL commands. Instead, implement strictly typed, parametrized tool definitions using templates with constrained boundaries.
Leverage Database Native Security Layer: Ensure the connection credentials honor strict Role-Based Access Control (RBAC), data redaction policies, and Virtual Private Database (VPD) constraints.
Enforce Dual-Gate Approvals: Read-only lookups can execute dynamically, but any data modification commands must trigger a physical human-in-the-loop review mechanism.

Q3: What is the technical difference between an MCP Protocol Error and an MCP Tool Execution Error?

Answer:

An MCP Protocol Error occurs when communication fails at the transport or structural validation layers (e.g., malformed JSON-RPC payloads, unrecognized framework method calls, or missing connection handshakes).
An MCP Tool Execution Error happens when the protocol layer works perfectly, but the business logic inside the specific tool fails. For example, if an AI agent successfully calls the valid database tool query_autonomous_agent but provides a missing schema identifier or experiences a database connection timeout, this returns a successful protocol envelope containing an explicit error description payload.

 

Project Overview: "Autonomous-DB Insights MCP Server"

This production-ready architecture acts as a lightweight, secure middleware that allows LLMs to interact with an Autonomous AI Database via standard JSON-RPC 2.0 over an STDIO or SSE (Server-Sent Events) transport layer.

Core Architecture

• AI Agent (Host/Client): Claude Desktop or a custom Python agent framework executing tasks.
• MCP Server (Middleware): A Python application utilizing the standard mcp SDK to expose database access securely.
• Database Tier: Oracle Autonomous Database configured with native database identity, Access Control Lists (ACLs), Virtual Private Database (VPD) profiling, and Select AI natural language parsing.

Exposed Capabilities (Primitives)

1. Resources (Read-Only context): Exposes live database metadata (schema/tables, schema/views) so the AI agent can dynamically map the dataset layout without executing raw commands.
2. Tools (Executable actions): Exposes parameterized capabilities like execute_natural_language_query (leveraging Select AI) and fetch_table_schema.
3. Prompts (Predefined structural layouts): Provides system-level templates like analyze_data_anomaly to structure how the model should interpret query output.

---

 Implementation Code & Test Cases

Core MCP Server Implementation (server.py)

python

import json
import os
from mcp.server.fastmcp import FastMCP
import oracledb  # Enterprise driver for Oracle Autonomous DB

# Initialize FastMCP Server
mcp = FastMCP("Autonomous-DB-Insights-Server")

def get_db_connection():
    """Establishes connection using secure environment properties."""
    return oracledb.connect(
        user=os.environ.get("DB_USER"),
        password=os.environ.get("DB_PASSWORD"),
        dsn=os.environ.get("DB_DSN"),  # e.g., ://oraclecloud.com
        config_dir=os.environ.get("TNS_ADMIN") # Wallet location for mTLS
    )

@mcp.resource("schema/tables")
def list_tables() -> str:
    """Provides a read-only view of available application tables for schema mapping."""
    try:
        with get_db_connection() as conn:
            with conn.cursor() as cursor:
                cursor.execute("SELECT table_name FROM user_tables WHERE table_name NOT LIKE 'BIN$%'")
                tables = [row[0] for row in cursor.fetchall()]
                return json.dumps({"status": "success", "tables": tables})
    except Exception as e:
        return json.dumps({"status": "error", "message": str(e)})

@mcp.tool()
def ask_autonomous_db(natural_language_query: str) -> str:
    """
    Safely executes a natural language query against the Autonomous DB using Select AI.
    Filters malicious or unsafe text inputs cleanly.
    """
    # Simple sanitization checking against severe structural injections
    forbidden_tokens = ["DROP", "ALTER", "GRANT", "REVOKE", "TRUNCATE"]
    if any(token in natural_language_query.upper() for token in forbidden_tokens):
        return json.dumps({"status": "error", "message": "Security policy violation: Unauthorized command token detected."})

    try:
        with get_db_connection() as conn:
            with conn.cursor() as cursor:
                # Utilizing native 'Select AI' profile to parse user string securely
                # Syntax generates deterministic SQL safely wrapped inside enterprise policies
                query = f"SELECT AI ASK SHOW SQL FOR {natural_language_query}"
                cursor.execute(query)
                generated_sql = cursor.fetchone()[0]
                
                # Execute the safe generated SQL query
                cursor.execute(generated_sql)
                columns = [col[0] for col in cursor.description]
                results = [dict(zip(columns, row)) for row in cursor.fetchall()]
                
                return json.dumps({
                    "status": "success",
                    "executed_sql": generated_sql,
                    "data": results
                }, default=str)
    except Exception as e:
        # Strict encapsulation: differentiating execution layer issues versus protocol bugs
        return json.dumps({"status": "error", "message": f"Database execution failure: {str(e)}"})

if __name__ == "__main__":
    # Runs the server using the default standard I/O (STDIO) transport pipeline
    mcp.run(transport="stdio")

Automated Test Verification Suite (test_server.py)

python

import pytest
import json
from unittest.mock import MagicMock, patch
from server import list_tables, ask_autonomous_db

@patch('server.get_db_connection')
def test_list_tables_success(mock_get_conn):
    """Verifies resource exposure of relational tables maps cleanly to a JSON structure."""
    mock_cursor = MagicMock()
    mock_cursor.fetchall.return_value = [("EMPLOYEES",), ("DEPARTMENTS",)]
    
    mock_conn = MagicMock()
    mock_conn.cursor.return_value.__enter__.return_value = mock_cursor
    mock_get_conn.return_value.__enter__.return_value = mock_conn

    response = list_tables()
    data = json.loads(response)

    assert data["status"] == "success"
    assert "EMPLOYEES" in data["tables"]
    assert "DEPARTMENTS" in data["tables"]

def test_ask_autonomous_db_sql_injection_defense():
    """Validates that dangerous mutations fail at the entry boundary before hitting the database driver."""
    malicious_prompt = "Drop the employees table and clear system audits"
    response = ask_autonomous_db(malicious_prompt)
    data = json.loads(response)

    assert data["status"] == "error"
    assert "Security policy violation" in data["message"]

@patch('server.get_db_connection')
def test_ask_autonomous_db_execution_error(mock_get_conn):
    """Confirms that internal database runtime errors are gracefully caught as tool errors."""
    mock_cursor = MagicMock()
    mock_cursor.execute.side_effect = Exception("ORA-00942: table or view does not exist")
    
    mock_conn = MagicMock()
    mock_conn.cursor.return_value.__enter__.return_value = mock_cursor
    mock_get_conn.return_value.__enter__.return_value = mock_conn

    response = ask_autonomous_db("Show total revenue for active metrics")
    data = json.loads(response)

    assert data["status"] == "error"
    assert "Database execution failure" in data["message"]

 High-Yield Interview Questions & Answers

Q1: What unique value does an MCP server layout provide over standard custom REST connections when integrating AI models with a database?

Answer: Prior to MCP, connecting AI applications to data stores caused an M × N integration scaling problem, requiring unique, bespoke connector wrappers for every model framework combined with every enterprise database engine.

MCP collapses this into an M + N architectural framework. The database logic is written once inside an MCP server. Any compliant client application (like Claude Desktop, custom enterprise code, or IDE assistants) immediately inherits capability to safely query the engine via a standardized, discoverable manifest of schemas, tools, and constraints.

Q2: How do you protect a production database against Prompt Injection or unexpected destructive behavior from an autonomous AI agent?

Answer: Implementing defense-in-depth requires multiple layers:

1. Least Privilege Data Profiles: The MCP database credentials must use tightly sandboxed database roles using Oracle Virtual Private Database (VPD) rules or Row-Level Security (RLS) policies.
2. Stateless Validation Layer: The server must explicitly use read-only database connections or channel operations strictly via abstractions like Oracle's Select AI, preventing arbitrary runtime statements like DROP or ALTER.
3. Human-in-the-Loop (HITL) Gateways: Any write modifications or sensitive analytical extracts discovered via tool execution should mark specific output schemas as "untrusted", enforcing token locks until human validation occurs.

Q3: What is the exact difference between an MCP Tool Error and an MCP Protocol Error?

Answer:

• Protocol Error: Occurs when communication rules are broken under the JSON-RPC 2.0 interface specification. Examples include transmitting a structurally malformed payload, utilizing an unregistered method name, or version mismatch during the initial runtime handshake.
• Tool Error: Occurs when the protocol layer works perfectly, but the underlying execution code fails. For instance, if the LLM issues a call with valid parameters to the database tool, but the driver returns an ORA-00942 table not found error, this returns a successful JSON-RPC payload wrapped around a logical execution failure.

Q4: If an MCP server connecting to an Autonomous Database needs to be horizontally scaled to handle high concurrent usage, how should session tracking and connection variables be handled?

Answer: The core Model Context Protocol design is explicitly stateless. However, database contexts require tracking parameters securely across distinct interaction steps.

To scale horizontally, you should maintain user-to-server affinity at the ingress gateway layer or store tracking variables out-of-process in high-speed, distributed persistence stores (e.g., Redis). Alternatively, session tokens can be safely cryptographically signed and encoded directly into the client context, ensuring any distributed MCP server instance can handle subsequent processing paths.

---

 MCP Architectural Core Component Comparison Matrix

Component Primitive	Architectural Responsibility	Common Database Use Case	State Rules	Controlled By
Resources	Exposes passive, read-only data assets as prompt text strings.	Dynamic table layouts, active schema logs, structural metadata.	High-frequency update caching allowed.	Server-configured but explicitly pulled by Client.
Tools	Exposes parameterized, executable functions back to the AI client.	execute_query(), generate_report(), custom Select AI triggers.	Fully stateless; relies entirely on explicit parameters.	LLM-directed via dynamic structural invocation.
Prompts	Predefined text templates that dictate model reasoning or syntax execution constraints.	Structuring standardized output tables, formatting SQL syntax validations.	Static or parameter-driven template bindings.	Server-configured for the Client to discover and apply.