[오라클] MCP를 통해 SELECT AI기능 사용하기

pip install fastmcp oracledb

# Import the FastMCP class from the fastmcp module to create an MCP server instance.
from fastmcp import FastMCP

# Import the oracledb module to enable connections to Oracle databases.
import oracledb

# Initialize the FastMCP server with a descriptive name.
mcp = FastMCP("Oracle Database Select AI", dependencies=["oracledb"])

# Define Oracle database connection parameters.
oracle_user = '<DB사용자명>'  # Username for the Oracle database.
oracle_pw = '<DB패스워드>'  # Password for the Oracle database user.
oracle_dsn = "tcps://<dns_name>:1522/<service_name>"  # Data Source Name specifying the Oracle database location.
oracle_wallet_location = '<wallet_location>'  # Path to the Oracle wallet directory.
oracle_wallet_pw = '<wallet_passwd>'  # Password for the Oracle wallet.

@mcp.tool()
def ask_profile_list() -> str:
    """
    Retrieves a list of AI profiles from the Oracle database.

    Connects to the Oracle database using the provided credentials and wallet information.
    Executes a SQL query to fetch all AI profiles, aggregating them into a JSON array.
    Returns:
        A JSON-formatted string containing profile names and their descriptions.
        If no profiles are found, returns a message indicating the absence of profiles.
    """
    # Establish a connection to the Oracle database using a context manager for automatic resource management.
    with oracledb.connect(
        user=oracle_user,
        password=oracle_pw,
        dsn=oracle_dsn,
        wallet_location=oracle_wallet_location,
        wallet_password=oracle_wallet_pw
    ) as conn:
        # Create a cursor object to execute SQL queries.
        with conn.cursor() as cur:
            # Define the SQL query to aggregate profile names and descriptions into a JSON array.
            sql = """
                SELECT JSON_ARRAYAGG(JSON_OBJECT('profile_name' VALUE profile_name, 'description' VALUE description))
                FROM user_cloud_ai_profiles where description is not null and status = 'ENABLED'
            """
            # Execute the SQL query without additional parameters.
            cur.execute(sql)
            # Fetch the first row of the result set.
            result = cur.fetchone()
    # Return the JSON array if profiles are found; otherwise, return a message indicating no profiles exist.
    return result[0] if result else "There is no profile."

@mcp.tool()
def ask_database(user_question: str, profile_name: str) -> str:
    """
    Sends a user question to Oracle Select AI and retrieves the generated response.

    Parameters:
        user_question (str): The question posed by the user to be processed by Select AI.
        profile_name (str): The name of the AI profile to be used for generating the response.

    Returns:
        str: The response generated by Oracle Select AI.
             If no response is generated, returns a message indicating the lack of response.
    """
    # Establish a connection to the Oracle database using a context manager.
    with oracledb.connect(
        user=oracle_user,
        password=oracle_pw,
        dsn=oracle_dsn,
        wallet_location=oracle_wallet_location,
        wallet_password=oracle_wallet_pw
    ) as conn:
        # Create a cursor object to execute SQL queries.
        with conn.cursor() as cur:
            # Define the SQL query to invoke the dbms_cloud_ai.generate function with the user's question and profile name.
            sql = """
                SELECT  json_serialize(dbms_cloud_ai.generate(
                    prompt => :1,
                    action => 'runsql',
                    profile_name => :2
                ) returning CLOB) 
                FROM DUAL
            """
            # Execute the SQL query with the user_question and profile_name as parameters.
            cur.execute(sql, [user_question, profile_name])
            # Fetch the first row of the result set.
            result = cur.fetchone()
            clob_text = result[0].read()
    # Return the generated response if available; otherwise, return a message indicating no response.
    return clob_text if result else "No response from Select AI."

# Entry point of the script.
if __name__ == "__main__":
    # Print a message indicating that the MCP server is starting.
    print("Starting MCP server...")
    # Run the MCP server to start handling requests.
    mcp.run(transport="stdio")