memora

Instructions

Implementation Reference

• memora/server.py:361-527 (handler)

  The `memory_create` tool handler implementation in `memora/server.py` which manages memory creation, metadata redaction, hierarchy path checking, similar memory suggestions, and automated tagging/hierarchical organization.

  @mcp.tool()
  async def memory_create(
      content: str,
      metadata: Optional[Dict[str, Any]] = None,
      tags: Optional[list[str]] = None,
      suggest_similar: bool = True,
      similarity_threshold: float = 0.2,
      response_mode: Literal["full", "minimal"] = "full",
  ) -> Dict[str, Any]:
      """Create a new memory entry.
  
      Args:
          content: The memory content text
          metadata: Optional metadata dictionary
          tags: Optional list of tags
          suggest_similar: If True, find similar memories and suggest consolidation (default: True)
          similarity_threshold: Minimum similarity score for suggestions (default: 0.2)
          response_mode: "full" (default) or "minimal" response payload size
      """
      if response_mode not in CREATE_RESPONSE_MODES:
          valid = ", ".join(sorted(CREATE_RESPONSE_MODES))
          return {
              "error": "invalid_input",
              "message": f"response_mode must be one of: {valid}",
          }
  
      # Check hierarchy path BEFORE creating to detect new paths
      new_path = extract_hierarchy_path(metadata)
      existing_paths = (
          get_existing_hierarchy_paths(
              _list_memories(None, None, None, 0, None, None, None, None, None)
          )
          if new_path
          else []
      )
      path_is_new = bool(new_path) and (new_path not in existing_paths)
  
      # Initialize warnings dict
      warnings: Dict[str, Any] = {}
  
      # Auto-redact secrets/PII from content BEFORE saving
      redacted_content = content.strip()
      try:
          redacted_content, secrets_redacted = _redact_secrets(redacted_content)
          if secrets_redacted:
              warnings["secrets_redacted"] = secrets_redacted
      except Exception as exc:
          logger.warning("Secret redaction failed, storing original content: %s", exc)
  
      try:
          record = _create_memory(content=redacted_content, metadata=metadata, tags=tags or [])
      except ValueError as exc:
          return {"error": "invalid_input", "message": str(exc)}
  
      result: Dict[str, Any] = {"memory": record}
  
      # Warn if a new hierarchy path was created and suggest similar existing paths
      if path_is_new:
          similar = find_similar_paths(new_path, existing_paths)
          if similar:
              warnings["new_hierarchy_path"] = f"New hierarchy path created: {new_path}"
              result["existing_similar_paths"] = similar
              result["hint"] = "Did you mean to use one of these existing paths? Use memory_update to change if needed."
  
      # Use cross-refs (related memories) for consolidation hints and duplicate detection
      # Cross-refs use full embedding context (content + metadata + tags) so are more accurate
      related_memories = record.get("related", []) if record else []
      if suggest_similar and related_memories:
          # Filter by threshold
          above_threshold = [m for m in related_memories if m and m.get("score", 0) >= similarity_threshold]
          if above_threshold:
              result["similar_memories"] = above_threshold
              result["consolidation_hint"] = (
                  f"Found {len(above_threshold)} similar memories. "
                  "Consider: (1) merge content with memory_update, or (2) delete redundant ones with memory_delete."
              )
              # Check for potential duplicates (>0.85 similarity)
              duplicates = [m for m in above_threshold if m.get("score", 0) >= DUPLICATE_THRESHOLD]
              if duplicates:
                  warnings["duplicate_warning"] = (
                      f"Very similar memory exists (>={int(DUPLICATE_THRESHOLD*100)}% match). "
                      f"Memory #{duplicates[0]['id']} has {int(duplicates[0]['score']*100)}% similarity."
                  )
  
      # Add warnings to result if any
      if warnings:
          result["warnings"] = warnings
  
      # Infer type and suggest tags (only if user didn't provide tags)
      try:
          suggestions: Dict[str, Any] = {}
          inferred_type = _infer_type(redacted_content)
          if inferred_type:
              suggestions["type"] = inferred_type
  
          suggested_tags = _suggest_tags(redacted_content, inferred_type)
          # Only suggest tags not already applied
          existing_tags = set(tags or [])
          new_suggestions = [t for t in suggested_tags if t not in existing_tags]
          if new_suggestions:
              suggestions["tags"] = new_suggestions
  
          # Suggest hierarchy placement based on related memories (cross-refs)
          # (only if user didn't provide a hierarchy path)
          if not new_path and related_memories:
              hierarchy_suggestions = suggest_hierarchy_from_similar(
                  related_memories,
                  get_memory_by_id=_get_memory,
              )
              if hierarchy_suggestions:
                  top = hierarchy_suggestions[0]
                  if top.get("confidence", 0) >= AUTO_HIERARCHY_THRESHOLD:
                      # Auto-apply the top hierarchy suggestion
                      auto_meta = {}
                      if top.get("section"):
                          auto_meta["section"] = top["section"]
                      if top.get("subsection"):
                          auto_meta["subsection"] = top["subsection"]
                      if auto_meta:
                          memory_id = record.get("id") if record else None
                          if memory_id is not None:
                              _update_memory(memory_id, None, auto_meta, None)
                              result["auto_hierarchy"] = {
                                  "path": top["path"],
                                  "section": top.get("section"),
                                  "subsection": top.get("subsection"),
                                  "confidence": top["confidence"],
                                  "source_memory_ids": top.get("similar_memory_ids", []),
                              }
                              result["auto_hierarchy_hint"] = (
                                  f"Auto-assigned hierarchy '{'/'.join(top['path'])}' "
                                  f"(confidence: {top['confidence']}) based on similar memories. "
                                  "Use memory_update to change if needed."
                              )
                  else:
                      # Below threshold — suggest but don't apply
                      suggestions["hierarchy"] = hierarchy_suggestions
                      suggestions["hierarchy_hint"] = (
                          "Similar memories are organized under these paths. "
                          "Use memory_update to add section/subsection metadata."
                      )
  
          if suggestions:
              result["suggestions"] = suggestions
      except Exception as exc:
          logger.warning(
              "Memory suggestion pipeline failed for memory id=%s: %s",
              (record or {}).get("id"),
              exc,
          )
  
      if response_mode == "minimal":
          minimal_result: Dict[str, Any] = {"memory": {"id": result["memory"]["id"]}}
          if "similar_memories" in result:
              minimal_result["similar_memories"] = result["similar_memories"]
          if "consolidation_hint" in result:
              minimal_result["consolidation_hint"] = result["consolidation_hint"]
          duplicate_warning = result.get("warnings", {}).get("duplicate_warning")
          if duplicate_warning:
              minimal_result["warnings"] = {"duplicate_warning": duplicate_warning}
  
          _schedule_cloud_graph_sync()
          return minimal_result
  
      _schedule_cloud_graph_sync()
      return result