diff --git a/src/mcp_cordra/client.py b/src/mcp_cordra/client.py index 17b71f3..938c4d1 100644 --- a/src/mcp_cordra/client.py +++ b/src/mcp_cordra/client.py @@ -78,3 +78,32 @@ class CordraClient: raise CordraNotFoundError(f"Object not found: {object_id}") from e raise CordraClientError(f"Failed to retrieve object {object_id}: {e}") from e + async def find(self, query: str) -> list[dict[str, Any]]: + """Find objects using a Cordra query. + + Args: + query: The query string to search for objects + + Returns: + List of objects matching the query as dictionaries + + Raises: + CordraClientError: If there's an API error + """ + try: + # Use CordraPy to find objects + # TODO - need to handle pagination, but the CordraPy API does not support it. + response: dict[str, Any] = cordra.CordraObject.find( + self.config.cordra_url, # type: ignore + query + ) + + # Extract the results array from the response + if isinstance(response, dict) and 'results' in response: + return response['results'] + else: + return [] + + except Exception as e: + raise CordraClientError(f"Failed to search with query '{query}': {e}") from e + diff --git a/src/mcp_cordra/server.py b/src/mcp_cordra/server.py index 192bd23..fe92ca0 100644 --- a/src/mcp_cordra/server.py +++ b/src/mcp_cordra/server.py @@ -42,6 +42,36 @@ async def get_cordra_object(prefix: str, suffix: str) -> str: raise RuntimeError(f"Failed to retrieve object {object_id}: {e}") from e +@mcp.resource("cordra://schemas", name="cordra-schemas-list", description="List available Cordra type schemas") +async def list_cordra_schemas() -> str: + """List available Cordra type schemas. + + Returns: + JSON array of available schema names + + Raises: + RuntimeError: If there's an API error + """ + try: + # Use the client's find method to get all schema objects + schemas = await cordra_client.find("type:Schema") + + # Extract the names from the schema objects + schema_names = [] + for schema in schemas: + if isinstance(schema, dict) and 'name' in schema: + schema_names.append(schema['name']) + + result = { + "schemas": schema_names, + "count": len(schema_names) + } + return json.dumps(result, indent=2) + + except Exception as e: + raise RuntimeError(f"Failed to list schemas: {e}") from e + + @mcp.tool() async def ping() -> str: """Simple ping tool to test server connectivity.""" diff --git a/tests/test_client.py b/tests/test_client.py index 945e832..d129f2c 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -156,6 +156,66 @@ class TestCordraClient: assert "Failed to retrieve object test/123" in str(exc_info.value) + @patch('mcp_cordra.client.cordra.CordraObject.find') + async def test_find_success(self, mock_find, client): + """Test successful find operation.""" + mock_response = { + "results": [ + {"name": "User", "identifier": "test/user-schema"}, + {"name": "Project", "identifier": "test/project-schema"}, + {"name": "Document", "identifier": "test/doc-schema"} + ], + "size": 3 + } + mock_find.return_value = mock_response + + result = await client.find("type:Schema") + + assert len(result) == 3 + assert result[0]["name"] == "User" + assert result[1]["name"] == "Project" + assert result[2]["name"] == "Document" + + mock_find.assert_called_once_with( + client.config.cordra_url, + "type:Schema" + ) + + @patch('mcp_cordra.client.cordra.CordraObject.find') + async def test_find_empty_results(self, mock_find, client): + """Test find with empty results.""" + mock_response = {"results": [], "size": 0} + mock_find.return_value = mock_response + + result = await client.find("type:NonExistent") + + assert result == [] + mock_find.assert_called_once_with( + client.config.cordra_url, + "type:NonExistent" + ) + + @patch('mcp_cordra.client.cordra.CordraObject.find') + async def test_find_no_results_key(self, mock_find, client): + """Test find with response missing results key.""" + mock_response = {"size": 0} # No results key + mock_find.return_value = mock_response + + result = await client.find("type:Schema") + + assert result == [] + + @patch('mcp_cordra.client.cordra.CordraObject.find') + async def test_find_error(self, mock_find, client): + """Test find error handling.""" + mock_find.side_effect = Exception("Search failed") + + with pytest.raises(CordraClientError) as exc_info: + await client.find("invalid:query") + + assert "Failed to search with query 'invalid:query'" in str(exc_info.value) + assert "Search failed" in str(exc_info.value) + class TestCordraConfig: """Test the CordraConfig class.""" diff --git a/tests/test_server.py b/tests/test_server.py index 4b9efe6..fe056b4 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -142,4 +142,106 @@ class TestGetCordraObject: assert parsed_result["content"]["id"] == "test/minimal" assert parsed_result["metadata"] is None assert parsed_result["acl"] is None - assert parsed_result["payloads"] is None \ No newline at end of file + assert parsed_result["payloads"] is None + + +class TestListCordraSchemas: + """Test the list_cordra_schemas resource handler.""" + + @patch('mcp_cordra.server.cordra_client') + async def test_list_schemas_success(self, mock_client): + """Test successful schema listing.""" + mock_schemas = [ + {"name": "User", "identifier": "test/user-schema"}, + {"name": "Project", "identifier": "test/project-schema"}, + {"name": "Document", "identifier": "test/doc-schema"}, + {"name": "CaptureEvent", "identifier": "test/capture-schema"} + ] + mock_client.find = AsyncMock(return_value=mock_schemas) + + from mcp_cordra.server import list_cordra_schemas + result = await list_cordra_schemas() + + # Verify the result is valid JSON + parsed_result = json.loads(result) + assert "schemas" in parsed_result + assert "count" in parsed_result + assert parsed_result["count"] == 4 + assert "User" in parsed_result["schemas"] + assert "Project" in parsed_result["schemas"] + assert "Document" in parsed_result["schemas"] + assert "CaptureEvent" in parsed_result["schemas"] + + # Verify the client was called with correct query + mock_client.find.assert_called_once_with("type:Schema") + + @patch('mcp_cordra.server.cordra_client') + async def test_list_schemas_empty(self, mock_client): + """Test schema listing with no results.""" + mock_client.find = AsyncMock(return_value=[]) + + from mcp_cordra.server import list_cordra_schemas + result = await list_cordra_schemas() + + parsed_result = json.loads(result) + assert parsed_result["schemas"] == [] + assert parsed_result["count"] == 0 + + mock_client.find.assert_called_once_with("type:Schema") + + @patch('mcp_cordra.server.cordra_client') + async def test_list_schemas_missing_name_field(self, mock_client): + """Test schema listing with objects missing name field.""" + mock_schemas = [ + {"name": "User", "identifier": "test/user-schema"}, + {"identifier": "test/no-name-schema"}, # Missing name field + {"name": "Project", "identifier": "test/project-schema"}, + {"other": "field"} # No name or identifier + ] + mock_client.find = AsyncMock(return_value=mock_schemas) + + from mcp_cordra.server import list_cordra_schemas + result = await list_cordra_schemas() + + parsed_result = json.loads(result) + assert parsed_result["count"] == 2 # Only objects with name field + assert "User" in parsed_result["schemas"] + assert "Project" in parsed_result["schemas"] + assert len(parsed_result["schemas"]) == 2 + + @patch('mcp_cordra.server.cordra_client') + async def test_list_schemas_client_error(self, mock_client): + """Test schema listing with client error.""" + from mcp_cordra.client import CordraClientError + mock_client.find = AsyncMock(side_effect=CordraClientError("Search failed")) + + from mcp_cordra.server import list_cordra_schemas + with pytest.raises(RuntimeError) as exc_info: + await list_cordra_schemas() + + assert "Failed to list schemas" in str(exc_info.value) + assert "Search failed" in str(exc_info.value) + + @patch('mcp_cordra.server.cordra_client') + async def test_list_schemas_json_format(self, mock_client): + """Test that the returned JSON is properly formatted.""" + mock_schemas = [ + {"name": "TestSchema", "identifier": "test/schema"} + ] + mock_client.find = AsyncMock(return_value=mock_schemas) + + from mcp_cordra.server import list_cordra_schemas + result = await list_cordra_schemas() + + # Verify it's valid JSON with proper indentation + parsed_result = json.loads(result) + assert isinstance(parsed_result, dict) + + # Check that the result contains indentation (pretty-printed) + assert " " in result # Should have 2-space indentation + + # Verify expected structure + assert "schemas" in parsed_result + assert "count" in parsed_result + assert isinstance(parsed_result["schemas"], list) + assert isinstance(parsed_result["count"], int) \ No newline at end of file