bauer/cordra-mcp
1
0
Fork 0
mirror of https://github.com/dnlbauer/cordra-mcp.git synced 2026-09-10 21:55:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: remove cordra object resource and made design object retrieval a tool call

Browse Source
This commit is contained in:
Daniel Bauer
2025-12-04 13:38:18 +01:00
parent d42bb1a581
commit ad6dce005b
3 changed files with 21 additions and 172 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
README.md
View File

@@ -20,9 +20,7 @@ ensuring safe exploration without risk of data modification or corruption.
### Resources
- `cordra://objects/{prefix}/{suffix}` - Retrieve a specific object by its handle identifier
- `cordra://schemas/{schema_name}` - Schema definition for a specific type.
- `cordra://design` - Design document containing the overall structure and configuration of the Cordra repository.
### Tools
@@ -36,6 +34,9 @@ ensuring safe exploration without risk of data modification or corruption.
- `count_objects` - Count the total number of objects matching a query.
- `query` - Lucene/Solr compatible search query
- `type` - Optional filter by object type
- `get_design_object` - Retrieve the Cordra design object containing repository configuration.
- Includes type definitions, workflow configurations, and system settings
- Administrative privileges are typically required to access this object
#### Query Syntax

55
src/cordra_mcp/server.py
View File

@@ -169,51 +169,18 @@ async def get_object(object_id: str) -> str:
raise RuntimeError(f"Failed to retrieve object {object_id}: {e}") from e
@mcp.resource(
"cordra://objects/{prefix}/{suffix}",
name="cordra-object",
title="Retrieve Cordra Digital Object",
description="Retrieve a Digital Object and Metadata from Cordra by its ID/handle.",
mime_type="application/json",
@mcp.tool(
name="get_design_object",
title="Get Cordra Design Object",
description="""
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: The design object as JSON
""",
)
async def get_cordra_object(prefix: str, suffix: str) -> str:
"""Retrieve a Cordra digital object by its ID.
Args:
prefix: The prefix part of the object ID (e.g., 'wildlive')
suffix: The suffix part of the object ID (e.g., '7a4b7b65f8bb155ad36d')
Returns:
JSON representation of the digital object
Raises:
RuntimeError: If the object is not found or there's an API error
"""
object_id = f"{prefix}/{suffix}"
try:
digital_object = await cordra_client.get_object(object_id)
object_dict = digital_object.model_dump()
return json.dumps(object_dict, indent=2)
except ValueError as e:
raise RuntimeError(f"Invalid parameters: {e}") from e
except CordraNotFoundError as e:
raise RuntimeError(f"Object not found: {object_id}") from e
except CordraAuthenticationError as e:
raise RuntimeError(f"Authentication failed: {e}") from e
except CordraClientError as e:
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:
async def get_cordra_design_object() -> str:
"""Retrieve the Cordra design object containing repository configuration.
The design object is the central location where Cordra stores its configuration,

133
tests/test_server.py
View File

@@ -14,8 +14,7 @@ from cordra_mcp.client import (
)
from cordra_mcp.server import (
count_objects,
get_cordra_design,
get_cordra_object,
get_cordra_design_object,
get_object,
search_objects,
)
@@ -45,124 +44,6 @@ def sample_digital_object() -> DigitalObject:
)
class TestGetCordraObject:
"""Test the get_cordra_object resource handler."""
@patch("cordra_mcp.server.cordra_client")
async def test_get_object_success(
self, mock_client: Any, sample_digital_object: DigitalObject
) -> None:
"""Test successful object retrieval."""
mock_client.get_object = AsyncMock(return_value=sample_digital_object)
result = await get_cordra_object("people", "john-doe-123")
# Verify the result is valid JSON
parsed_result = json.loads(result)
assert parsed_result["id"] == "people/john-doe-123"
assert parsed_result["type"] == "Person"
assert parsed_result["content"]["name"] == "John Doe"
assert parsed_result["content"]["birthday"] == "1990-05-15"
assert parsed_result["metadata"]["created"] == "2023-01-01"
assert len(parsed_result["payloads"]) == 1
assert parsed_result["payloads"][0]["name"] == "profile_photo"
# Verify the client was called with the correct object ID
mock_client.get_object.assert_called_once_with("people/john-doe-123")
@patch("cordra_mcp.server.cordra_client")
async def test_get_object_not_found(self, mock_client: Any) -> None:
"""Test object not found exception."""
mock_client.get_object = AsyncMock(
side_effect=CordraNotFoundError("Object not found: people/nonexistent")
)
with pytest.raises(RuntimeError) as exc_info:
await get_cordra_object("people", "nonexistent")
assert "Object not found: people/nonexistent" in str(exc_info.value)
mock_client.get_object.assert_called_once_with("people/nonexistent")
@patch("cordra_mcp.server.cordra_client")
async def test_get_object_client_error(self, mock_client: Any) -> None:
"""Test general client error handling."""
mock_client.get_object = AsyncMock(
side_effect=CordraClientError("Connection failed")
)
with pytest.raises(RuntimeError) as exc_info:
await get_cordra_object("people", "john-doe-123")
assert "Failed to retrieve object people/john-doe-123" in str(exc_info.value)
assert "Connection failed" in str(exc_info.value)
mock_client.get_object.assert_called_once_with("people/john-doe-123")
@patch("cordra_mcp.server.cordra_client")
async def test_object_id_construction(
self, mock_client: Any, sample_digital_object: DigitalObject
) -> None:
"""Test that object ID is correctly constructed from prefix and suffix."""
mock_client.get_object = AsyncMock(return_value=sample_digital_object)
# Test various prefix/suffix combinations
test_cases = [
("people", "john-doe-123", "people/john-doe-123"),
("documents", "report-2023", "documents/report-2023"),
("items", "item_with_underscores", "items/item_with_underscores"),
]
for prefix, suffix, expected_id in test_cases:
await get_cordra_object(prefix, suffix)
mock_client.get_object.assert_called_with(expected_id)
@patch("cordra_mcp.server.cordra_client")
async def test_json_formatting(
self, mock_client: Any, sample_digital_object: DigitalObject
) -> None:
"""Test that the returned JSON is properly formatted."""
mock_client.get_object = AsyncMock(return_value=sample_digital_object)
result = await get_cordra_object("people", "john-doe-123")
# 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
assert "acl" in parsed_result
assert "payloads" in parsed_result
@patch("cordra_mcp.server.cordra_client")
async def test_minimal_object(self, mock_client: Any) -> None:
"""Test handling of object with minimal data."""
minimal_object = DigitalObject(
id="test/minimal",
type="",
content={"id": "test/minimal"},
metadata=None,
acl=None,
payloads=None,
)
mock_client.get_object = AsyncMock(return_value=minimal_object)
result = await get_cordra_object("test", "minimal")
parsed_result = json.loads(result)
assert parsed_result["id"] == "test/minimal"
assert parsed_result["type"] == ""
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
class TestGetObject:
"""Test the get_object tool."""
@@ -663,7 +544,7 @@ class TestSearchObjects:
class TestGetCordraDesign:
"""Test the get_cordra_design resource handler."""
"""Test the get_cordra_design_object tool."""
@patch("cordra_mcp.server.cordra_client")
async def test_get_design_success(self, mock_client: Any) -> None:
@@ -680,7 +561,7 @@ class TestGetCordraDesign:
)
mock_client.get_design = AsyncMock(return_value=mock_design)
result = await get_cordra_design()
result = await get_cordra_design_object()
# Verify the result is valid JSON
parsed_result = json.loads(result)
@@ -701,7 +582,7 @@ class TestGetCordraDesign:
)
with pytest.raises(RuntimeError) as exc_info:
await get_cordra_design()
await get_cordra_design_object()
assert "Design object not found" in str(exc_info.value)
mock_client.get_design.assert_called_once()
@@ -714,7 +595,7 @@ class TestGetCordraDesign:
)
with pytest.raises(RuntimeError) as exc_info:
await get_cordra_design()
await get_cordra_design_object()
assert "Authentication failed" in str(exc_info.value)
mock_client.get_design.assert_called_once()
@@ -727,7 +608,7 @@ class TestGetCordraDesign:
)
with pytest.raises(RuntimeError) as exc_info:
await get_cordra_design()
await get_cordra_design_object()
assert "Failed to retrieve design object" in str(exc_info.value)
assert "Connection failed" in str(exc_info.value)
@@ -744,7 +625,7 @@ class TestGetCordraDesign:
)
mock_client.get_design = AsyncMock(return_value=mock_design)
result = await get_cordra_design()
result = await get_cordra_design_object()
# Verify it's valid JSON with proper indentation
parsed_result = json.loads(result)