feat: add get_object tool for retrieving Cordra digital objects by ID with error handling

This commit is contained in:
Daniel Bauer
2025-12-02 17:34:45 +01:00
parent 0a9ba538ef
commit 3c8367f804
3 changed files with 99 additions and 0 deletions

View File

@@ -26,6 +26,8 @@ ensuring safe exploration without risk of data modification or corruption.
### Tools
- `get_object` - Retrieve a digital object by its complete ID/handle.
- `object_id` - Complete object ID (e.g., "test/abc123")
- `search_objects` - Search for digital objects using a query string with pagination support.
- `query` - Lucene/Solr compatible search query
- `type` - Optional filter by object type

View File

@@ -134,6 +134,41 @@ async def count_objects(
raise RuntimeError(f"Count failed: {e}") from e
@mcp.tool(
name="get_object",
title="Get Cordra Object by ID",
description="""Retrieve a digital object by its complete ID/handle.
Returns: Full object with metadata as JSON
Example: get_object("test/abc123")""",
)
async def get_object(object_id: str) -> str:
"""Retrieve a Cordra digital object by its complete ID.
Args:
object_id: The complete object ID/handle (e.g., "test/abc123" or "wildlive/7a4b7b65f8bb155ad36d")
Returns:
JSON string containing the complete digital object with all metadata
Raises:
RuntimeError: If the object is not found or there's an API error
"""
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 object ID: {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://objects/{prefix}/{suffix}",
name="cordra-object",

View File

@@ -16,6 +16,7 @@ from cordra_mcp.server import (
count_objects,
get_cordra_design,
get_cordra_object,
get_object,
search_objects,
)
@@ -162,6 +163,67 @@ class TestGetCordraObject:
assert parsed_result["payloads"] is None
class TestGetObject:
"""Test the get_object tool."""
@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 with complete ID."""
mock_client.get_object = AsyncMock(return_value=sample_digital_object)
result = await get_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"
# 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: test/nonexistent")
)
with pytest.raises(RuntimeError) as exc_info:
await get_object("test/nonexistent")
assert "Object not found: test/nonexistent" in str(exc_info.value)
mock_client.get_object.assert_called_once_with("test/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_object("test/obj123")
assert "Failed to retrieve object test/obj123" in str(exc_info.value)
mock_client.get_object.assert_called_once_with("test/obj123")
@patch("cordra_mcp.server.cordra_client")
async def test_get_object_authentication_error(self, mock_client: Any) -> None:
"""Test authentication error handling."""
mock_client.get_object = AsyncMock(
side_effect=CordraAuthenticationError("Authentication failed")
)
with pytest.raises(RuntimeError) as exc_info:
await get_object("test/obj123")
assert "Authentication failed" in str(exc_info.value)
mock_client.get_object.assert_called_once_with("test/obj123")
class TestSchemaResourceFunctions:
"""Test the schema resource functions."""