Skip to content

fix some tools not returning a string #3

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Aug 10, 2025
Merged

fix some tools not returning a string #3

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 9 additions & 10 deletions src/dbtools-mcp-server/dbtools-mcp-server.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"""
Expand All @@ -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."})

Expand All @@ -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:
Expand All @@ -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:
Expand All @@ -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:
Expand All @@ -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 = []
Expand All @@ -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:
Expand All @@ -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):
"""
Expand Down
43 changes: 14 additions & 29 deletions src/dbtools-mcp-server/test_dbtools_mcp_server.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"""
Expand All @@ -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")
Expand All @@ -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")

Expand Down