diff --git a/src/dbtools-mcp-server/dbtools-mcp-server.py b/src/dbtools-mcp-server/dbtools-mcp-server.py index d8a86e2..27db3b6 100755 --- a/src/dbtools-mcp-server/dbtools-mcp-server.py +++ b/src/dbtools-mcp-server/dbtools-mcp-server.py @@ -44,8 +44,7 @@ def list_all_compartments() -> str: """List all compartments in a tenancy with clear formatting""" compartments = identity_client.list_compartments(tenancy_id).data compartments.append(identity_client.get_compartment(compartment_id=tenancy_id).data) - - return compartments + return str(compartments) def get_compartment_by_name(compartment_name: str): """Internal function to get compartment by name with caching""" @@ -69,7 +68,7 @@ def get_compartment_by_name_tool(name: str) -> str: """Return a compartment matching the provided name""" compartment = get_compartment_by_name(name) if compartment: - return compartment + return str(compartment) else: return json.dumps({"error": f"Compartment '{name}' not found."}) @@ -81,7 +80,7 @@ def list_autonomous_databases(compartment_name: str) -> str: return json.dumps({"error": f"Compartment '{compartment_name}' not found. Use list_compartment_names() to see available compartments."}) databases = database_client.list_autonomous_databases(compartment_id=compartment.id).data - return databases + return str(databases) @mcp.tool() def list_all_databases() -> str: @@ -92,7 +91,7 @@ def list_all_databases() -> str: matching_context_type="NONE" ) results = search_client.search_resources(search_details=search_details, tenant_id=config['tenancy']).data - return results + return str(results) @mcp.tool() def list_dbtools_connection_tool(compartment_name: str) -> str: @@ -102,7 +101,7 @@ def list_dbtools_connection_tool(compartment_name: str) -> str: return json.dumps({"error": f"Compartment '{compartment_name}' not found. Use list_compartment_names() to see available compartments."}) connections = dbtools_client.list_database_tools_connections(compartment_id=compartment.id).data - return connections + return str(connections) @mcp.tool() def list_all_connections() -> str: @@ -115,7 +114,7 @@ def list_all_connections() -> str: search_results = search_client.search_resources(search_details=search_details, tenant_id=config['tenancy']).data if not hasattr(search_results, 'items'): - return [] + return json.dumps([]) # Get full details for each connection detailed_results = [] @@ -127,10 +126,10 @@ def list_all_connections() -> str: # If we can't get details for a connection, include error info detailed_results.append({ "error": f"Error getting details for connection {item.display_name}: {str(e)}", - "search_result": item + "search_result": item.identifier }) - return detailed_results + return str(detailed_results) @mcp.tool() def get_dbtools_connection_by_name_tool(display_name: str) -> str: @@ -153,7 +152,7 @@ def get_dbtools_connection_by_name_tool(display_name: str) -> str: # Get the full connection details connection = dbtools_client.get_database_tools_connection(connection_id).data - return connection + return str(connection) def get_minimal_connection_by_name(dbtools_connection_display_name: str): """ diff --git a/src/dbtools-mcp-server/test_dbtools_mcp_server.py b/src/dbtools-mcp-server/test_dbtools_mcp_server.py index dbc12e7..18d68e0 100755 --- a/src/dbtools-mcp-server/test_dbtools_mcp_server.py +++ b/src/dbtools-mcp-server/test_dbtools_mcp_server.py @@ -47,8 +47,8 @@ def setUpClass(cls): cls.module = cls.server_module # Specific connection names for testing - cls.oracle_connection = "oracleconn1" - cls.mysql_connection = "mysqlconn1" + cls.oracle_connection = os.getenv("ORACLE_CONNECTION_NAME", "oracleconn1") + cls.mysql_connection = os.getenv("MYSQL_CONNECTION_NAME", "mysqlconn1") def setUp(self): """Set up test case - verify OCI config exists""" @@ -62,32 +62,15 @@ def setUp(self): print(f"Running test: {self._testMethodName}") print(f"{'=' * 70}") - def test_get_help(self): - """Test the get_help tool""" - print("About to call get_help() to retrieve tool documentation") - result = self.module.get_help() - - # Parse result as JSON - help_data = json.loads(result) - - # Verify expected keys in the help text - self.assertIn("overview", help_data) - self.assertIn("recommended_workflow", help_data) - self.assertIn("available_tools", help_data) - self.assertIn("examples", help_data) - - # Print summary of what we got - print(f"Help overview: {help_data['overview']}") - print(f"Found {len(help_data['recommended_workflow'])} workflow steps") - print(f"Found {len(help_data['available_tools'])} tool categories") - def test_list_all_compartments(self): """Test listing all compartments""" print("About to call list_all_compartments() to list all compartments in the tenancy") - result = self.module.list_all_compartments() + str_result = self.module.list_all_compartments() # Verify we got a list of compartments - self.assertIsNotNone(result) + self.assertIsNotNone(str_result) + + result = json.loads(str_result) # Print how many compartments we found print(f"Found {len(result)} compartments") @@ -96,28 +79,30 @@ def test_list_all_compartments(self): if len(result) > 0: print("First few compartments:") for i, comp in enumerate(result[:3]): - print(f" {i+1}. {comp.name} (ID: {comp.id})") + print(f" {i+1}. {comp['name']} (ID: {comp['id']})") def test_get_compartment_by_name(self): """Test getting a compartment by name""" # First get all compartments print("About to call list_all_compartments() to find a compartment name for testing") - all_compartments = self.module.list_all_compartments() + all_compartments_str = self.module.list_all_compartments() + all_compartments = json.loads(all_compartments_str) # If we have any compartments, test with the first one's name if len(all_compartments) > 0: first_compartment = all_compartments[0] - compartment_name = first_compartment.name + print(first_compartment) + compartment_name = first_compartment['name'] print(f"About to call get_compartment_by_name_tool('{compartment_name}')") # Now try to get this compartment by name - result = self.module.get_compartment_by_name_tool(compartment_name) + result = json.loads(self.module.get_compartment_by_name_tool(compartment_name)) # Verify we got a result self.assertIsNotNone(result) - self.assertEqual(result.name, compartment_name) + self.assertEqual(result['name'], compartment_name) - print(f"Successfully retrieved compartment: {result.name} (ID: {result.id})") + print(f"Successfully retrieved compartment: {result['name']} (ID: {result['id']})") else: self.skipTest("No compartments found to test with")