feat: add schema listing functionality

- Add find method to CordraClient for querying objects
- Implement cordra://schemas resource for listing available type schemas
- Add comprehensive unit tests for both client find method and server schema listing
- Handle CordraPy response format with results array extraction
- Support schema name extraction from returned objects
This commit is contained in:
daniel
2025-06-29 14:28:08 +02:00
parent 0f9d33c20b
commit cc06e1eace
4 changed files with 222 additions and 1 deletions

View File

@@ -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

View File

@@ -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."""