feat: add design object resource for repository configuration access

Add new MCP resource cordra://design to retrieve Cordra's central design object
containing repository configuration, type definitions, and system settings.

- Add CordraClient.get_design() method using /api/objects/design endpoint
- Add get_cordra_design() resource handler with proper MCP annotations
- Include comprehensive error handling for authentication failures
- Document administrative privilege requirements in descriptions
- Add complete test coverage for both client and server functionality

The design object provides AI systems access to understand the complete
data model and configuration structure of a Cordra repository.
This commit is contained in:
Daniel Bauer
2025-07-04 10:07:16 +02:00
parent 94340e2bb1
commit 748d9b604f
4 changed files with 245 additions and 2 deletions

View File

@@ -213,3 +213,44 @@ class CordraClient:
raise CordraClientError(
f"Failed to retrieve schema '{schema_name}': {e}"
) from e
async def get_design(self) -> DigitalObject:
"""Retrieve the Cordra design object containing repository configuration.
The design object contains the central configuration for the Cordra repository
including type definitions, workflow configurations, and system settings.
Administrative privileges are typically required to access this object.
Returns:
The design object as a DigitalObject
Raises:
CordraNotFoundError: If the design object is not found
CordraAuthenticationError: If authentication fails or insufficient privileges
CordraClientError: For other API errors
"""
url = f"{self.config.base_url}/api/objects/design"
try:
response = self.session.get(url, timeout=self.config.timeout)
if not response.ok:
self._handle_http_error(
response, "Failed to retrieve design object"
)
design_obj = response.json()
return DigitalObject(
id="design",
type=design_obj.get("type", "CordraDesign"),
content=design_obj.get("content", design_obj),
metadata=design_obj.get("metadata"),
acl=design_obj.get("acl"),
payloads=design_obj.get("payloads"),
)
except requests.RequestException as e:
raise CordraClientError(
f"Failed to retrieve design object: {e}"
) from e

View File

@@ -105,6 +105,39 @@ 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://design",
name="cordra-design",
title="Retrieve Cordra Design Object",
description="Retrieve the Cordra design object containing repository configuration. Administrative privileges are typically required to access this object.",
mime_type="application/json",
)
async def get_cordra_design() -> str:
"""Retrieve the Cordra design object containing repository configuration.
The design object is the central location where Cordra stores its configuration,
including type definitions, workflow configurations, and system settings.
Administrative privileges are typically required to access this object.
Returns:
JSON representation of the design object
Raises:
RuntimeError: If the design object is not found, authentication fails, or there's an API error
"""
try:
design_object = await cordra_client.get_design()
object_dict = design_object.model_dump()
return json.dumps(object_dict, indent=2)
except CordraNotFoundError as e:
raise RuntimeError("Design object not found") from e
except CordraAuthenticationError as e:
raise RuntimeError(f"Authentication failed: {e}") from e
except CordraClientError as e:
raise RuntimeError(f"Failed to retrieve design object: {e}") from e
async def create_schema_resource(schema_name: str) -> str:
"""Create content for a specific schema resource."""
try:

View File

@@ -5,6 +5,7 @@ from unittest.mock import patch
import pytest
from cordra_mcp.client import (
CordraAuthenticationError,
CordraClient,
CordraClientError,
CordraNotFoundError,
@@ -296,6 +297,71 @@ class TestCordraClient:
timeout=30,
)
@patch("cordra_mcp.client.requests.Session.get")
async def test_get_design_success(self, mock_get, client):
"""Test successful design object retrieval."""
mock_design_data = {
"type": "CordraDesign",
"content": {
"types": {"User": {}, "Project": {}},
"workflows": {},
"systemConfig": {"serverName": "test-cordra"}
},
"metadata": {"created": "2023-01-01", "modified": "2023-06-15"}
}
mock_response = mock_get.return_value
mock_response.status_code = 200
mock_response.json.return_value = mock_design_data
mock_response.ok = True
result = await client.get_design()
assert isinstance(result, DigitalObject)
assert result.id == "design"
assert result.type == "CordraDesign"
assert result.content["systemConfig"]["serverName"] == "test-cordra"
mock_get.assert_called_once_with(
"https://test.example.com/api/objects/design",
timeout=30,
)
@patch("cordra_mcp.client.requests.Session.get")
async def test_get_design_authentication_error(self, mock_get, client):
"""Test design object retrieval with authentication error."""
mock_response = mock_get.return_value
mock_response.status_code = 403
mock_response.ok = False
with pytest.raises(CordraAuthenticationError) as exc_info:
await client.get_design()
assert "Authentication failed" in str(exc_info.value)
@patch("cordra_mcp.client.requests.Session.get")
async def test_get_design_not_found(self, mock_get, client):
"""Test design object retrieval with not found error."""
mock_response = mock_get.return_value
mock_response.status_code = 404
mock_response.ok = False
with pytest.raises(CordraNotFoundError) as exc_info:
await client.get_design()
assert "Resource not found" in str(exc_info.value)
@patch("cordra_mcp.client.requests.Session.get")
async def test_get_design_request_error(self, mock_get, client):
"""Test design object retrieval with request error."""
from requests import RequestException
mock_get.side_effect = RequestException("Connection failed")
with pytest.raises(CordraClientError) as exc_info:
await client.get_design()
assert "Failed to retrieve design object" in str(exc_info.value)
class TestCordraConfig:
"""Test the CordraConfig class."""

View File

@@ -5,8 +5,13 @@ from unittest.mock import AsyncMock, patch
import pytest
from cordra_mcp.client import CordraClientError, CordraNotFoundError, DigitalObject
from cordra_mcp.server import get_cordra_object, search_objects
from cordra_mcp.client import (
CordraAuthenticationError,
CordraClientError,
CordraNotFoundError,
DigitalObject,
)
from cordra_mcp.server import get_cordra_design, get_cordra_object, search_objects
@pytest.fixture
@@ -380,3 +385,101 @@ class TestSearchObjects:
assert parsed_result[0]["id"] == "test/object"
assert parsed_result[0]["type"] == "Test"
assert parsed_result[0]["content"]["data"] == "value"
class TestGetCordraDesign:
"""Test the get_cordra_design resource handler."""
@patch('cordra_mcp.server.cordra_client')
async def test_get_design_success(self, mock_client):
"""Test successful design object retrieval."""
mock_design = DigitalObject(
id="design",
type="CordraDesign",
content={
"types": {"User": {}, "Project": {}},
"workflows": {},
"systemConfig": {"serverName": "test-cordra"}
},
metadata={"created": "2023-01-01", "modified": "2023-06-15"}
)
mock_client.get_design = AsyncMock(return_value=mock_design)
result = await get_cordra_design()
# Verify the result is valid JSON
parsed_result = json.loads(result)
assert parsed_result["id"] == "design"
assert parsed_result["type"] == "CordraDesign"
assert parsed_result["content"]["systemConfig"]["serverName"] == "test-cordra"
assert "types" in parsed_result["content"]
assert "workflows" in parsed_result["content"]
# Verify the client was called
mock_client.get_design.assert_called_once()
@patch('cordra_mcp.server.cordra_client')
async def test_get_design_not_found(self, mock_client):
"""Test design object not found exception."""
mock_client.get_design = AsyncMock(
side_effect=CordraNotFoundError("Design object not found")
)
with pytest.raises(RuntimeError) as exc_info:
await get_cordra_design()
assert "Design object not found" in str(exc_info.value)
mock_client.get_design.assert_called_once()
@patch('cordra_mcp.server.cordra_client')
async def test_get_design_authentication_error(self, mock_client):
"""Test design object authentication error."""
mock_client.get_design = AsyncMock(
side_effect=CordraAuthenticationError("Authentication failed")
)
with pytest.raises(RuntimeError) as exc_info:
await get_cordra_design()
assert "Authentication failed" in str(exc_info.value)
mock_client.get_design.assert_called_once()
@patch('cordra_mcp.server.cordra_client')
async def test_get_design_client_error(self, mock_client):
"""Test design object general client error."""
mock_client.get_design = AsyncMock(
side_effect=CordraClientError("Connection failed")
)
with pytest.raises(RuntimeError) as exc_info:
await get_cordra_design()
assert "Failed to retrieve design object" in str(exc_info.value)
assert "Connection failed" in str(exc_info.value)
mock_client.get_design.assert_called_once()
@patch('cordra_mcp.server.cordra_client')
async def test_get_design_json_formatting(self, mock_client):
"""Test that the design object is properly formatted as JSON."""
mock_design = DigitalObject(
id="design",
type="CordraDesign",
content={"data": "value"},
metadata={"created": "2023-01-01"}
)
mock_client.get_design = AsyncMock(return_value=mock_design)
result = await get_cordra_design()
# 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 all expected fields are present
assert "id" in parsed_result
assert "type" in parsed_result
assert "content" in parsed_result
assert "metadata" in parsed_result