feat: add search tool for digital object discovery

Implement MCP tool to search Cordra repository using Lucene/Solr syntax.
Enables AI assistants to discover and filter digital objects by content,
type, and other metadata fields.

Features:
- Lucene/Solr compatible query syntax support
- Optional type filtering (e.g., Person, Document)
- Configurable result limits with sensible defaults
- Comprehensive error handling and validation
- JSON formatted results with proper indentation

Technical changes:
- Enhanced CordraClient.find() with type and limit parameters
- Added search_objects MCP tool with detailed annotations
- Removed unused tools module in favor of decorator approach
- Added 12 comprehensive unit tests covering all scenarios
This commit is contained in:
Daniel Bauer
2025-07-04 09:30:38 +02:00
parent 46e1a8057f
commit 94340e2bb1
5 changed files with 273 additions and 5 deletions

View File

@@ -129,11 +129,13 @@ class CordraClient:
f"Failed to retrieve object {object_id}: {e}"
) from e
async def find(self, query: str) -> list[dict[str, Any]]:
async def find(self, query: str, object_type: str | None = None, limit: int | None = None) -> list[dict[str, Any]]:
"""Find objects using a Cordra query.
Args:
query: The query string to search for objects
object_type: Optional filter by object type
limit: Optional limit on number of results
Returns:
List of objects matching the query as dictionaries
@@ -143,15 +145,24 @@ class CordraClient:
CordraAuthenticationError: If authentication fails
CordraClientError: For other API errors
"""
# Construct the final query with type filter if specified
final_query = query
if object_type:
final_query = f"type:{object_type} AND ({query})"
url = f"{self.config.base_url}/search"
params = {"query": query}
params = {"query": final_query}
# Add pageSize if limit is specified
if limit is not None:
params["pageSize"] = str(limit)
try:
response = self.session.get(url, params=params, timeout=self.config.timeout)
if not response.ok:
self._handle_http_error(
response, f"Failed to search with query '{query}'"
response, f"Failed to search with query '{final_query}'"
)
search_result = response.json()
@@ -164,7 +175,7 @@ class CordraClient:
except requests.RequestException as e:
raise CordraClientError(
f"Failed to search with query '{query}': {e}"
f"Failed to search with query '{final_query}': {e}"
) from e
async def get_schema(self, schema_name: str) -> DigitalObject: