fix: update project name to 'cordra-mcp' in configuration files

src/cordra_mcp/__init__.py

@@ -0,0 +1,3 @@
+"""MCP server for Cordra digital object repository."""
+
+__version__ = "0.1.0"

src/cordra_mcp/client.py

@@ -0,0 +1,185 @@
+"""Cordra client wrapper using HTTP requests."""
+
+import logging
+from typing import Any
+
+import requests
+from pydantic import BaseModel, Field
+
+from .config import CordraConfig
+
+logger = logging.getLogger(__name__)
+
+
+class DigitalObject(BaseModel):
+    """Model for a Cordra digital object."""
+
+    id: str = Field(description="Object identifier")
+    type: str = Field(description="Object type")
+    content: dict[str, Any] = Field(description="Object content as JSON")
+    metadata: dict[str, Any] | None = Field(default=None, description="Object metadata")
+    acl: dict[str, Any] | None = Field(default=None, description="Access control list")
+    payloads: list[dict[str, Any]] | None = Field(default=None, description="List of payloads")
+
+
+class CordraClientError(Exception):
+    """Base exception for Cordra client errors."""
+    pass
+
+
+class CordraNotFoundError(CordraClientError):
+    """Exception raised when an object is not found."""
+    pass
+
+
+class CordraAuthenticationError(CordraClientError):
+    """Exception raised for authentication/authorization failures."""
+    pass
+
+
+class CordraClient:
+    """Client for interacting with Cordra repository using HTTP requests."""
+
+    def __init__(self, config: CordraConfig) -> None:
+        """Initialize the Cordra client.
+
+        Args:
+            config: Configuration settings for the Cordra connection
+        """
+        self.config = config
+        self.session = requests.Session()
+        self.session.verify = config.verify_ssl
+
+        # Set up authentication
+        if config.username and config.password:
+            self.session.auth = (config.username, config.password)
+        elif config.username or config.password:
+            logger.warning("Only username or password provided, not both. Authentication may fail.")
+
+    def _handle_http_error(self, response: requests.Response, context: str) -> None:
+        """Handle HTTP errors and raise appropriate exceptions.
+
+        Args:
+            response: The HTTP response object
+            context: Context description for the error message
+
+        Raises:
+            CordraNotFoundError: For 404 errors
+            CordraAuthenticationError: For 401/403 errors
+            CordraClientError: For other HTTP errors
+        """
+        status_code = response.status_code
+        if status_code == 404:
+            raise CordraNotFoundError(f"{context}: Resource not found")
+        elif status_code in (401, 403):
+            raise CordraAuthenticationError(f"{context}: Authentication failed (HTTP {status_code})")
+        elif status_code >= 500:
+            raise CordraClientError(f"{context}: Server error (HTTP {status_code})")
+        else:
+            raise CordraClientError(f"{context}: HTTP error {status_code}")
+
+    async def get_object(self, object_id: str) -> DigitalObject:
+        """Retrieve a digital object by its ID.
+
+        Args:
+            object_id: The unique identifier of the object to retrieve
+
+        Returns:
+            The full digital object
+
+        Raises:
+            ValueError: If object_id is empty
+            CordraNotFoundError: If the object is not found
+            CordraAuthenticationError: If authentication fails
+            CordraClientError: For other API errors
+        """
+        url = f"{self.config.cordra_url}/objects/{object_id}"
+        params = {"full": "true"}
+
+        try:
+            response = self.session.get(url, params=params, timeout=self.config.timeout)
+
+            if not response.ok:
+                self._handle_http_error(response, f"Failed to retrieve object {object_id}")
+
+            cordra_obj = response.json()
+
+            return DigitalObject(
+                id=object_id,
+                type=cordra_obj.get('type', ''),
+                content=cordra_obj.get('content', cordra_obj),
+                metadata=cordra_obj.get('metadata'),
+                acl=cordra_obj.get('acl'),
+                payloads=cordra_obj.get('payloads'),
+            )
+
+        except requests.RequestException as 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:
+            ValueError: If query is empty
+            CordraAuthenticationError: If authentication fails
+            CordraClientError: For other API errors
+        """
+        url = f"{self.config.cordra_url}/search"
+        params = {"query": query}
+
+        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}'")
+
+            search_result = response.json()
+
+            # Extract the results array from the response
+            if isinstance(search_result, dict) and 'results' in search_result:
+                return search_result['results']
+            else:
+                return []
+
+        except requests.RequestException as e:
+            raise CordraClientError(f"Failed to search with query '{query}': {e}") from e
+
+    async def get_schema(self, schema_name: str) -> DigitalObject:
+        """Retrieve a schema definition by its name.
+
+        Args:
+            schema_name: The name of the schema to retrieve
+
+        Returns:
+            The schema object containing the type definition
+
+        Raises:
+            CordraNotFoundError: If the schema is not found
+            CordraAuthenticationError: If authentication fails
+            CordraClientError: For other API errors
+        """
+        # Search for the specific schema by name using correct query format
+        query = f"type:Schema AND /name:{schema_name}"
+
+        try:
+            schemas = await self.find(query)
+
+            if not schemas:
+                raise CordraNotFoundError(f"Schema '{schema_name}' not found")
+
+            # Get the first matching schema (should be unique by name)
+            schema_data = schemas[0]
+
+            # Get the full schema object using its ID
+            return await self.get_object(schema_data['id'])
+
+        except (CordraNotFoundError, CordraAuthenticationError):
+            raise
+        except Exception as e:
+            raise CordraClientError(f"Failed to retrieve schema '{schema_name}': {e}") from e

src/cordra_mcp/config.py

@@ -0,0 +1,43 @@
+"""Configuration settings for the MCP Cordra server."""
+
+
+from pydantic import Field
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class CordraConfig(BaseSettings):
+    """Configuration for connecting to a Cordra repository."""
+
+    model_config = SettingsConfigDict(
+        env_prefix="CORDRA_",
+        env_file=".env",
+        env_file_encoding="utf-8",
+        case_sensitive=False,
+    )
+
+    cordra_url: str = Field(
+        default="https://localhost:8443",
+        description="Base URL of the Cordra repository"
+    )
+    username: str | None = Field(
+        default=None,
+        description="Username for Cordra authentication"
+    )
+    password: str | None = Field(
+        default=None,
+        description="Password for Cordra authentication"
+    )
+    max_search_results: int = Field(
+        default=1000,
+        description="Maximum number of search results to return"
+    )
+    verify_ssl: bool = Field(
+        default=True,
+        description="Whether to verify SSL certificates"
+    )
+    timeout: int = Field(
+        default=30,
+        description="Request timeout in seconds"
+    )
+
+

src/cordra_mcp/server.py

@@ -0,0 +1,129 @@
+"""MCP server for Cordra digital object repository."""
+
+import asyncio
+import json
+import logging
+
+from mcp.server.fastmcp import FastMCP
+from mcp.server.fastmcp.resources import FunctionResource
+
+from .client import (
+    CordraAuthenticationError,
+    CordraClient,
+    CordraClientError,
+    CordraNotFoundError,
+)
+from .config import CordraConfig
+
+# Initialize the MCP server
+mcp = FastMCP("cordra-mcp")
+
+# Initialize Cordra client at startup
+config = CordraConfig()
+cordra_client = CordraClient(config)
+
+logger = logging.getLogger(__name__)
+
+
+@mcp.resource(
+    "cordra://objects/{prefix}/{suffix}",
+    name="cordra-object",
+    description="Retrieve a Cordra digital object by ID",
+)
+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
+
+
+async def create_schema_resource(schema_name: str) -> str:
+    """Create content for a specific schema resource."""
+    try:
+        schema_object = await cordra_client.get_schema(schema_name)
+        schema_dict = schema_object.model_dump()
+        return json.dumps(schema_dict, indent=2)
+    except CordraNotFoundError as e:
+        raise RuntimeError(f"Schema not found: {schema_name}") from e
+    except CordraAuthenticationError as e:
+        raise RuntimeError(f"Authentication failed: {e}") from e
+    except CordraClientError as e:
+        raise RuntimeError(f"Failed to retrieve schema {schema_name}: {e}") from e
+
+
+async def register_schema_resources() -> None:
+    """Register individual schema resources dynamically."""
+    try:
+        # Get all available schemas
+        schemas = await cordra_client.find("type:Schema")
+
+        for schema in schemas:
+            schema_name = schema.get("content", {}).get("name")
+            if not schema_name:
+                logger.warning("Schema without a name found, skipping.")
+                continue
+
+            logger.info(f"Registering schema resource for cordra type {schema_name}")
+
+            async def schema_fn(name: str = schema_name) -> str:
+                return await create_schema_resource(name)
+
+            mcp.add_resource(
+                FunctionResource.from_function(
+                    uri=f"cordra://schemas/{schema_name}",
+                    fn=schema_fn,
+                    name=f"cordra-type-schema-{schema_name}",
+                    description=f"JSON schema for Cordra type {schema_name}",
+                )
+            )
+
+        logger.info(f"Registered {len(schemas)} schema resources")
+
+    except Exception as e:
+        logger.warning(f"Failed to register schema resources: {e}")
+
+
+@mcp.tool()
+async def ping() -> str:
+    """Simple ping tool to test server connectivity."""
+    return "pong"
+
+
+async def initialize_server() -> None:
+    """Initialize server resources before starting."""
+    logger.info("Initializing Cordra MCP server...")
+    await register_schema_resources()
+    logger.info("Server initialization complete")
+
+
+def main() -> None:
+    """Main entry point for the MCP server."""
+    asyncio.run(initialize_server())
+    mcp.run()
+
+
+if __name__ == "__main__":
+    main()