CEDAR MCP Server

Instructions

Implementation Reference

• src/cedar_mcp/server.py:92-186 (handler)

  The main MCP tool handler for get_instances_based_on_template, including registration via @mcp.tool(). Validates params, searches instance IDs, fetches and cleans each instance, handles pagination and errors.

  @mcp.tool()
  def get_instances_based_on_template(
      template_id: str, limit: int = 10, offset: int = 0
  ) -> Dict[str, Any]:
      """
      Get template instances that belong to the input template ID with pagination support.
  
      This tool searches for instances of a given template and fetches their complete content
      in paginated chunks to avoid token limit issues.
  
      Args:
          template_id: The template ID or full URL from CEDAR repository
                      (e.g., "https://repo.metadatacenter.org/templates/e019284e-48d1-4494-bc83-ddefd28dfbac")
          limit: Number of instances to return per page (min: 1, max: 100, default: 10)
          offset: Starting position for pagination (default: 0)
  
      Returns:
          Dictionary containing:
          - instances: List of template instances for this page
          - pagination: Pagination metadata (total_count, current_page, etc.)
          - errors: List of any errors encountered during fetching
      """
      # Validate pagination parameters
      if limit < 1 or limit > 100:
          return {
              "error": "Invalid limit parameter. Must be between 1 and 100.",
              "instances": [],
              "pagination": None,
          }
  
      if offset < 0:
          return {
              "error": "Invalid offset parameter. Must be 0 or greater.",
              "instances": [],
              "pagination": None,
          }
  
      # Step 1: Search for instance IDs with pagination
      search_result = search_instance_ids(
          template_id=template_id,
          cedar_api_key=CEDAR_API_KEY,
          limit=limit,
          offset=offset,
      )
  
      # Check if search failed
      if "error" in search_result:
          return {
              "error": f"Failed to search for template instances: {search_result['error']}",
              "instances": [],
              "pagination": None,
          }
  
      instance_ids = search_result.get("instance_ids", [])
      pagination_metadata = search_result.get("pagination", {})
  
      # If no instances found, return empty result with pagination metadata
      if not instance_ids:
          return {"instances": [], "pagination": pagination_metadata, "errors": None}
  
      # Step 2: Fetch content for each instance in this page
      instances = []
      failed_instances = []
  
      for instance_id in instance_ids:
          instance_content = get_instance(instance_id, CEDAR_API_KEY)
  
          # Check if this instance fetch failed
          if "error" in instance_content:
              failed_instances.append(
                  {"instance_id": instance_id, "error": instance_content["error"]}
              )
          else:
              # Clean the instance content before adding to results
              try:
                  cleaned_instance = clean_template_instance_response(
                      instance_content
                  )
                  instances.append(cleaned_instance)
              except Exception as e:
                  failed_instances.append(
                      {
                          "instance_id": instance_id,
                          "error": f"Failed to clean instance response: {str(e)}",
                      }
                  )
  
      # Step 3: Prepare response
      response = {"instances": instances, "pagination": pagination_metadata}
  
      # Include error information if any instances failed to fetch
      if failed_instances:
          response["errors"] = failed_instances
  
      return response

• src/cedar_mcp/external_api.py:56-135 (helper)

  Helper function that performs the paginated search on CEDAR API to find instance IDs based on template_id.

  def search_instance_ids(
      template_id: str, cedar_api_key: str, limit: int = 10, offset: int = 0
  ) -> Dict[str, Any]:
      """
      Search for template instances with pagination support.
  
      Args:
          template_id: Template ID (UUID or full URL)
          cedar_api_key: CEDAR API key for authentication
          limit: Number of instances to fetch
          offset: Starting position
  
      Returns:
          Dictionary containing:
          - instance_ids: List of instance IDs for this page
          - pagination: Pagination metadata
          - error: Error message if failed
      """
      try:
          # Convert template ID to full URL format if needed
          if not template_id.startswith("https://"):
              template_url = f"https://repo.metadatacenter.org/templates/{template_id}"
          else:
              template_url = template_id
  
          headers = {
              "Accept": "application/json",
              "Authorization": f"apiKey {cedar_api_key}",
          }
  
          # Build the search API URL
          base_url = "https://resource.metadatacenter.org/search"
          params = {
              "version": "latest",
              "limit": limit,
              "is_based_on": template_url,
              "offset": offset,
          }
  
          # Make the API request
          response = requests.get(
              base_url, headers=headers, params=cast(Any, params), timeout=30
          )
          response.raise_for_status()
          search_data = response.json()
  
          # Extract information
          total_count = search_data.get("totalCount", 0)
          resources = search_data.get("resources", [])
  
          # Extract instance IDs from current page
          instance_ids = []
          for resource in resources:
              instance_id = resource.get("@id")
              if instance_id:
                  instance_ids.append(instance_id)
  
          # Calculate pagination metadata
          current_page = (offset // limit) + 1
          total_pages = (
              (total_count + limit - 1) // limit if limit > 0 else 0
          )  # Ceiling division
          has_next = offset + limit < total_count
  
          pagination_metadata = {
              "limit": limit,
              "offset": offset,
              "total_count": total_count,
              "current_page": current_page,
              "total_pages": total_pages,
              "has_next": has_next,
          }
  
          return {"instance_ids": instance_ids, "pagination": pagination_metadata}
  
      except requests.exceptions.RequestException as e:
          return {"error": f"Failed to search CEDAR instances: {str(e)}"}
      except (KeyError, ValueError) as e:
          return {"error": f"Failed to parse CEDAR search response: {str(e)}"}

• src/cedar_mcp/external_api.py:137-172 (helper)

  Helper function that fetches the full content of a single template instance by its ID.

  def get_instance(instance_id: str, cedar_api_key: str) -> Dict[str, Any]:
      """
      Fetch the full content of a CEDAR template instance.
  
      Args:
          instance_id: Full instance URL (e.g., "https://repo.metadatacenter.org/template-instances/{uuid}")
          cedar_api_key: CEDAR API key for authentication
  
      Returns:
          Dictionary containing instance content or error information
      """
      try:
          # URL encode the instance ID for safe inclusion in URL
          encoded_instance_id = quote(instance_id, safe="")
  
          # Build the instance API URL
          base_url = f"https://resource.metadatacenter.org/template-instances/{encoded_instance_id}"
  
          # Set authorization header
          headers = {
              "Accept": "application/json",
              "Authorization": f"apiKey {cedar_api_key}",
          }
  
          # Make the API request with timeout
          response = requests.get(base_url, headers=headers, timeout=30)
          response.raise_for_status()
  
          # Return the raw JSON response from CEDAR
          return response.json()
  
      except requests.exceptions.RequestException as e:
          return {"error": f"Failed to fetch CEDAR instance content: {str(e)}"}
      except (KeyError, ValueError) as e:
          return {"error": f"Failed to parse CEDAR instance response: {str(e)}"}

• src/cedar_mcp/processing.py:454-492 (helper)

  Helper function that cleans and simplifies the raw JSON-LD response from a template instance.

  def clean_template_instance_response(instance_data: Dict[str, Any]) -> Dict[str, Any]:
      """
      Clean and transform CEDAR template instance JSON-LD to simplified structure.
  
      Removes metadata fields and transforms JSON-LD specific attributes:
      - Removes: '@context', 'schema:isBasedOn', 'schema:name', 'schema:description',
                 'pav:createdOn', 'pav:createdBy', 'pav:derivedFrom', 'oslc:modifiedBy', '@id' from root
      - Transforms: '@id' → 'iri', 'rdfs:label' → 'label' throughout
      - Flattens: '@value' objects to their direct values
  
      Args:
          instance_data: Raw instance data from CEDAR (JSON-LD format)
  
      Returns:
          Cleaned and transformed instance data as dictionary
      """
      # Remove metadata fields from root level
      metadata_fields = {
          "@context",
          "schema:isBasedOn",
          "schema:name",
          "schema:description",
          "pav:createdOn",
          "pav:createdBy",
          "pav:derivedFrom",
          "oslc:modifiedBy",
          "@id",
      }
  
      # Create cleaned copy
      cleaned_data = {}
      for key, value in instance_data.items():
          if key not in metadata_fields:
              cleaned_data[key] = value
  
      # Recursively transform the entire structure
      return _transform_jsonld_structure(cleaned_data)