Oracle MCP Server

Instructions

Implementation Reference

• src/oracle_mcp_server/server.py:465-517 (handler)

  The QueryExecutor.explain_query method implements the core tool logic: connects to Oracle DB, generates a unique statement ID, executes EXPLAIN PLAN, retrieves the hierarchical execution plan from plan_table with indentation, collects plan steps (operation, object, cost, cardinality, bytes), cleans up plan_table, and returns the structured execution plan.

  async def explain_query(self, sql: str) -> Dict[str, Any]:
      """Get execution plan for a query"""
      conn = await self.connection_manager.get_connection()
      try:
          cursor = conn.cursor()
  
          # Generate unique statement ID
          statement_id = f"MCP_EXPLAIN_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
  
          # Explain the plan
          explain_sql = f"EXPLAIN PLAN SET STATEMENT_ID = '{statement_id}' FOR {sql}"
          cursor.execute(explain_sql)
  
          # Fetch the execution plan
          plan_query = """
              SELECT 
                  LPAD(' ', 2 * (LEVEL - 1)) || operation || ' ' || options AS operation,
                  object_name,
                  cost,
                  cardinality,
                  bytes
              FROM plan_table
              WHERE statement_id = :statement_id
              START WITH id = 0
              CONNECT BY PRIOR id = parent_id AND statement_id = :statement_id
              ORDER BY id
          """
  
          cursor.execute(plan_query, [statement_id, statement_id])
  
          plan_rows = []
          for row in cursor:
              plan_rows.append(
                  {
                      "operation": row[0],
                      "object_name": row[1],
                      "cost": row[2],
                      "cardinality": row[3],
                      "bytes": row[4],
                  }
              )
  
          # Clean up
          cursor.execute(
              "DELETE FROM plan_table WHERE statement_id = :statement_id",
              [statement_id],
          )
          conn.commit()
  
          return {"execution_plan": plan_rows, "statement_id": statement_id}
  
      finally:
          conn.close()

• src/oracle_mcp_server/server.py:709-722 (registration)

  Registration of the explain_query tool in the MCP server's list_tools handler. Defines the tool name, description, and input schema requiring a 'sql' string parameter.

  Tool(
      name="explain_query",
      description="Get the execution plan for a SQL query to analyze performance",
      inputSchema={
          "type": "object",
          "properties": {
              "sql": {
                  "type": "string",
                  "description": "SQL query to explain",
              }
          },
          "required": ["sql"],
      },
  ),

• src/oracle_mcp_server/server.py:712-721 (schema)

  Input schema definition for the explain_query tool, specifying an object with a required 'sql' string property.

  inputSchema={
      "type": "object",
      "properties": {
          "sql": {
              "type": "string",
              "description": "SQL query to explain",
          }
      },
      "required": ["sql"],
  },

• src/oracle_mcp_server/server.py:835-843 (registration)

  Dispatch logic in the MCP server's call_tool handler that extracts 'sql' argument, calls the executor.explain_query, formats result as JSON text content, and returns it.

  elif name == "explain_query":
      sql = arguments.get("sql")
      result = await self.executor.explain_query(sql)
  
      return [
          TextContent(
              type="text", text=json.dumps(result, indent=2, default=str)
          )
      ]