Documentation Agent Skills

CopyDOCS_BASE = "https://docs.mem0.ai"
DOCS_HOST = urllib.parse.urlsplit(DOCS_BASE).netloc
LLMS_INDEX = f"{DOCS_BASE}/llms.txt"
ENTRY_URL = re.compile(r"\((https?://[^)\s]+)\)")
MAX_RESULTS = 20
MAX_PAGE_CHARS = 10000
MAX_INDEX_BYTES = 2_000_000
 
SECTION_PREFIXES = {
    "platform": ("/platform/",),
    "api": ("/api-reference/",),
    "open-source": ("/open-source/", "/components/"),
    "integrations": ("/integrations",),
}
 
 
def require_docs_url(url: str) -> None:
    """Exit unless the URL is an https URL on the Mem0 docs host."""
    parts = urllib.parse.urlsplit(url)
    if parts.scheme != "https" or parts.netloc != DOCS_HOST:
        sys.exit(f"Refusing to fetch {url}: only {DOCS_BASE} is allowed")
 
 
class DocsRedirectHandler(urllib.request.HTTPRedirectHandler):
    """Follow redirects only when they stay on the Mem0 docs host."""
 
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        require_docs_url(newurl)
        return super().redirect_request(req, fp, code, msg, headers, newurl)
 

CopyDOCS_BASE = "https://docs.mem0.ai"
DOCS_HOST = urllib.parse.urlsplit(DOCS_BASE).netloc
LLMS_INDEX = f"{DOCS_BASE}/llms.txt"
ENTRY_URL = re.compile(r"\((https?://[^)\s]+)\)")
MAX_RESULTS = 20
MAX_PAGE_CHARS = 10000
MAX_INDEX_BYTES = 2_000_000

Copydef require_docs_url(url: str) -> None:
    """Exit unless the URL is an https URL on the Mem0 docs host."""
    parts = urllib.parse.urlsplit(url)
    if parts.scheme != "https" or parts.netloc != DOCS_HOST:
        sys.exit(f"Refusing to fetch {url}: only {DOCS_BASE} is allowed")
 
 
class DocsRedirectHandler(urllib.request.HTTPRedirectHandler):
    """Follow redirects only when they stay on the Mem0 docs host."""
 
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        require_docs_url(newurl)
        return super().redirect_request(req, fp, code, msg, headers, newurl)

CopySECTION_PREFIXES = {
    "platform": ("/platform/",),
    "api": ("/api-reference/",),
    "open-source": ("/open-source/", "/components/"),
    "integrations": ("/integrations",),
}

CopySECTION_PREFIXES = {
    "platform": ("/platform/",),
    "api": ("/api-reference/",),
    "open-source": ("/open-source/", "/components/"),
    "integrations": ("/integrations",),
}

Copydef search_docs(query: str, section: str | None = None) -> dict:
    """Search the llms.txt index for entries matching the query terms, best matches first."""
    terms = re.findall(r"[a-z0-9_]+", query.lower())
    entries = index_entries()
 
    if section in SECTION_PREFIXES:
        entries = [e for e in entries if entry_path(e).startswith(SECTION_PREFIXES[section])]
 
    scored = []
    for entry in entries:
        haystack = entry.lower()
        title = haystack.split("]")[0]
        score = sum((term in haystack) + (term in title) for term in terms)
        if score:
            scored.append((score, entry))
    scored.sort(key=lambda item: -item[0])
 
    return {
        "source": "llms_txt_index",
        "query": query,
        "matching_urls": [entry for _, entry in scored[:MAX_RESULTS]],
        "suggestion": "Fetch specific pages with --page <path> for detailed content",
    }

Copydef index_entries() -> list:
    """Return the page entries listed in the llms.txt index."""
    content, truncated = fetch_url(LLMS_INDEX, MAX_INDEX_BYTES)
    if truncated:
        sys.exit(f"Error: {LLMS_INDEX} exceeds {MAX_INDEX_BYTES} bytes")
    return [line.strip()[2:] for line in content.splitlines() if line.strip().startswith("- [")]
 

Copyskills/mem0/scripts/mem0_doc_search.py:36| ENTRY_URL = re.compile(r"\((https?://[^)\s]+)\)")
skills/mem0/scripts/mem0_doc_search.py:87-91

Copycontent, truncated = fetch_url(LLMS_INDEX, MAX_INDEX_BYTES)
if truncated:
    sys.exit(f"Error: {LLMS_INDEX} exceeds {MAX_INDEX_BYTES} bytes")

Copydef fetch_page(page_path: str) -> dict:
    """Fetch a specific documentation page as markdown."""
    parts = urllib.parse.urlsplit(urllib.parse.urljoin(DOCS_BASE, page_path))
    path = parts.path.rstrip("/")
    if not path.endswith(".md"):
        path = f"{path}.md"
    url = urllib.parse.urlunsplit((parts.scheme, parts.netloc, path, "", ""))

Copyskills/mem0/scripts/mem0_doc_search.py:38| MAX_PAGE_CHARS = 10000
skills/mem0/scripts/mem0_doc_search.py:125-126

Copydef get_index() -> dict:
    """Fetch the full documentation index from llms.txt."""
    urls = index_entries()
    return {"total_pages": len(urls), "urls": urls, "sections": list(SECTION_PREFIXES)}
 
 
def list_section(section: str) -> dict:
    """List the llms.txt index entries in a documentation section."""
    if section not in SECTION_PREFIXES:
        return {"error": f"Unknown section: {section}", "available": list(SECTION_PREFIXES)}
    pages = [e for e in index_entries() if entry_path(e).startswith(SECTION_PREFIXES[section])]
    return {"section": section, "pages": pages}
 

Copyexport function loadSkill(
  skillName: string,
  config: SkillsConfig = {},
): LoadedSkill | null {
  const raw = readSkillFile(skillName);
  if (!raw) return null;
 
  const parsed = parseSkillFile(raw);
  const parts: string[] = [parsed.body];
 
  // Domain overlays only apply to the skill they target (checked via applies_to)
  if (config.domain) {
    const overlay = readDomainOverlay(config.domain, skillName);
    if (overlay) {
      parts.push("\n" + overlay);
    }
  }
 
  // Inject user-configured categories into triage skill prompt
  if (skillName === "memory-triage" && config.categories) {
    const mergedCats = resolveCategories(config);
    parts.push(renderCategoriesBlock(mergedCats));
  }
 
  // Inject triage knobs (importanceThreshold, credentialPatterns)
  if (skillName === "memory-triage") {
    const knobs = renderTriageKnobs(config);
    if (knobs) parts.push(knobs);
  }
 
  // Append user custom rules to the triage prompt.
  if (skillName === "memory-triage" && config.customRules) {
    const rulesBlock: string[] = ["\n## User Custom Rules\n"];
    if (config.customRules.include?.length) {
      rulesBlock.push("Additionally extract:");
      for (const rule of config.customRules.include) {
        rulesBlock.push(`- ${rule}`);
      }
    }
    if (config.customRules.exclude?.length) {
      rulesBlock.push("\nAdditionally skip:");
      for (const rule of config.customRules.exclude) {
        rulesBlock.push(`- ${rule}`);
      }
    }
    parts.push(rulesBlock.join("\n"));
  }
 
  return {
    name: skillName,
    prompt: parts.join("\n"),
    frontmatter: parsed.frontmatter,
  };
}

Copyif (config.domain) {
  const overlay = readDomainOverlay(config.domain, skillName);
  if (overlay) {
    parts.push("\n" + overlay);
  }
}

Copyfunction renderCompactCustomRules(config: SkillsConfig): string[] {
  const includeRules = config.customRules?.include ?? [];
  const excludeRules = config.customRules?.exclude ?? [];
  if (includeRules.length === 0 && excludeRules.length === 0) {
    return [];
  }
 
  const formatRuleList = (label: string, rules: string[]) =>
    `${label}: ${rules.map((rule) => `"${rule}"`).join("; ")}`;
 
  const lines: string[] = [];
  if (includeRules.length > 0) {
    lines.push(formatRuleList("Include", includeRules));
  }
  if (excludeRules.length > 0) {
    lines.push(formatRuleList("Exclude", excludeRules));
  }
 
  const combined = lines.join(" ");
  if (combined.length <= COMPACT_CUSTOM_RULE_CHAR_BUDGET) {
    return ["## Active Custom Rules", "", ...lines.map((line) => `- ${line}`)];
  }
 
  const preview = [
    ...includeRules.slice(0, COMPACT_CUSTOM_RULE_PREVIEW_LIMIT).map((rule) => `include "${rule}"`),
    ...excludeRules.slice(0, COMPACT_CUSTOM_RULE_PREVIEW_LIMIT).map((rule) => `exclude "${rule}"`),
  ];
 
  return [
    "## Active Custom Rules",
    "",
    `- ${includeRules.length} include rule(s) and ${excludeRules.length} exclude rule(s) are configured.`,
    `- Prompt kept compact, full rule text omitted because it exceeded ${COMPACT_CUSTOM_RULE_CHAR_BUDGET} characters.`,
    ...(preview.length > 0
      ? [`- Preview: ${preview.join("; ")}`]
      : []),
  ];
}

Copyconst COMPACT_CUSTOM_RULE_CHAR_BUDGET = 280;
const COMPACT_CUSTOM_RULE_PREVIEW_LIMIT = 2;
…
if (combined.length <= COMPACT_CUSTOM_RULE_CHAR_BUDGET) {
  return ["## Active Custom Rules", "", ...lines.map((line) => `- ${line}`)];
}
 
const preview = [
  ...includeRules.slice(0, COMPACT_CUSTOM_RULE_PREVIEW_LIMIT).map((rule) => `include "${rule}"`),
  ...excludeRules.slice(0, COMPACT_CUSTOM_RULE_PREVIEW_LIMIT).map((rule) => `exclude "${rule}"`),
];
 
return [
  "## Active Custom Rules",
  "",
  `- ${includeRules.length} include rule(s) and ${excludeRules.length} exclude rule(s) are configured.`,
  `- Prompt kept compact, full rule text omitted because it exceeded ${COMPACT_CUSTOM_RULE_CHAR_BUDGET} characters.`,
  ...(preview.length > 0
    ? [`- Preview: ${preview.join("; ")}`]
    : []),
];

Copyexport function loadTriagePrompt(config: SkillsConfig = {}): string {
  // Try to load the full skill with all config-driven overlays
…
let systemContext = loadCompactTriagePrompt(cfg.skills ?? {});
if (isSubagent) {
  systemContext =
    "You are a subagent — use these memories for context but do not assume you are this user. Do NOT store new memories.\n\n" +
    systemContext;
}
 
// Dynamic recall goes in prependContext (changes every turn).
// Strategy controls how much the plugin searches automatically:
//   "always" — long-term + session search every turn (2 searches)
//   "smart"  — long-term search only, no session search (1 search) [default]
//   "manual" — no auto-recall; agent controls all search via memory_search (0 searches)
let recallContext = "";
const recallEnabled = cfg.skills?.recall?.enabled !== false;
const recallStrategy = cfg.skills?.recall?.strategy ?? "smart";
 
if (recallEnabled && recallStrategy !== "manual") {
  const recallStart = Date.now();
  try {
    const query = lifecycle.prepareUserText(sanitizeQuery(event.prompt));
 
    // Smart mode: skip session search (saves 1 API call per turn)
    const sessionIdForRecall =
      recallStrategy === "always"
        ? isSubagent
          ? undefined
          : sessionId
        : undefined; // smart: long-term only
 
    const recallResult = await skillRecall(
      provider,
      query,
      userId,
      cfg.skills ?? {},
      sessionIdForRecall,
    );
 
    api.logger.info(
      `openclaw-mem0: skills-mode recall (strategy=${recallStrategy}) injecting ${recallResult.memories.length} memories (~${recallResult.tokenEstimate} tokens)`,
    );
 
    _captureEvent("openclaw.hook.recall", {
      strategy: recallStrategy,
      memory_count: recallResult.memories.length,
      latency_ms: Date.now() - recallStart,
    });
 
    recallContext = recallResult.context;
  } catch (err) {
    api.logger.warn(
      `openclaw-mem0: skills-mode recall failed: ${String(err)}`,
    );
  }
} else if (recallEnabled && recallStrategy === "manual") {
  api.logger.info(
    "openclaw-mem0: skills-mode recall strategy=manual, agent controls search",
  );
}
 
return {
  prependSystemContext: systemContext, // cached by provider
  prependContext: recallContext, // per-turn dynamic
};

Copy//   "always" — long-term + session search every turn (2 searches)
//   "smart"  — long-term search only, no session search (1 search) [default]
//   "manual" — no auto-recall; agent controls all search via memory_search (0 searches)
let recallContext = "";
…
const sessionIdForRecall =
  recallStrategy === "always"
    ? isSubagent
      ? undefined
      : sessionId
    : undefined; // smart: long-term only

Copydef load_ignore_prefixes() -> list[str]:
    if not IGNORE_FILE.exists():
        return []
    prefixes: list[str] = []
    for raw in IGNORE_FILE.read_text().splitlines():
        line = raw.strip()
        if line and not line.startswith("#"):
            prefixes.append(line)
    return prefixes
 
 
def canonical_repo_pages(ignore_prefixes: list[str]) -> tuple[set[str], set[str]]:
    all_pages: set[str] = set()
    included_pages: set[str] = set()
    for path in DOCS_DIR.rglob("*.mdx"):
        rel = path.relative_to(DOCS_DIR).with_suffix("").as_posix()
        all_pages.add(rel)
        if not any(rel.startswith(p) for p in ignore_prefixes):
            included_pages.add(rel)
    return all_pages, included_pages
 
 
def indexed_urls(text: str) -> set[str]:
    urls: set[str] = set()
    for match in URL_RE.finditer(text):
        path = match.group(1).rstrip("/")
        if path:
            urls.add(path)
    return urls

Copydefault   read-only; exits 1 if any drift is found.
--write   appends placeholder entries for missing pages under a
          "## Unclassified - needs triage" H2 at the end of
          docs/llms.txt. Stale URLs are reported but never removed
          automatically (human decides whether a page was renamed
          or genuinely deleted). Exits 1 if anything was written,
          0 if the file was already in sync.
…
missing = sorted(included_pages - linked)
stale = sorted(linked - all_pages)

Copydocs/llms.txt. Stale URLs are reported but never removed
automatically (human decides whether a page was renamed
or genuinely deleted). Exits 1 if anything was written,
…
print(
    "\n(Stale URLs are never auto-removed: check whether the page was "
    "renamed, and update the link by hand.)"
)

Copydef format_placeholder(page: str) -> str:
    title = page.rsplit("/", 1)[-1].replace("-", " ").replace("_", " ").title()
    return (
        f"- [{title}]({BASE_URL}{page}) [TODO: Platform|OSS|Both]: "
        "TODO - rewrite as 'Use when ...' and move into the correct section."
    )
 
 
def append_triage_block(text: str, missing_pages: list[str]) -> str:
    entries = "\n".join(format_placeholder(p) for p in missing_pages)
    if TRIAGE_HEADER in text:
        return text.rstrip() + "\n" + entries + "\n"
    preamble = (
        "> New docs pages were detected by CI. For each entry below: replace "
        "the scope tag with `[Platform]`, `[OSS]`, or `[Both]`; rewrite the "
        "description as `Use when ...`; move it into the correct section; "
        "then delete this H2 once empty.\n"
    )
    return text.rstrip() + "\n\n" + TRIAGE_HEADER + "\n\n" + preamble + "\n" + entries + "\n"
 
…
if args.write:
    if missing:
        new_text = append_triage_block(text, missing)
        LLMS_TXT.write_text(new_text)
        print(
            f"\nAppended {len(missing)} placeholder entries under "
            f"'{TRIAGE_HEADER}' in docs/llms.txt."
        )

Copydef format_placeholder(page: str) -> str:
    title = page.rsplit("/", 1)[-1].replace("-", " ").replace("_", " ").title()
    return (
        f"- [{title}]({BASE_URL}{page}) [TODO: Platform|OSS|Both]: "
        "TODO - rewrite as 'Use when ...' and move into the correct section."
    )

CopyPROTOCOL_VERSION = "2024-11-05"
TOOL_NAME = "search_memories"
TOOL_DESCRIPTION = (
    "Search memories from earlier work in this repository. Use it before "
    "repeating investigation or when earlier decisions, fixes, commands, or "
    "results may help."
)
…
def handle_request(message: Any) -> dict[str, Any] | None:
    if not isinstance(message, dict):
        return None
    request_id = message.get("id")
    method = message.get("method")
 
    if method == "notifications/initialized":
        return None
    if method == "initialize":
        requested = (message.get("params") or {}).get("protocolVersion")
        return {
            "jsonrpc": "2.0",
            "id": request_id,
            "result": {
                "protocolVersion": requested or PROTOCOL_VERSION,
                "capabilities": {"tools": {"listChanged": False}},
                "serverInfo": {"name": "mem0", "version": PLUGIN_VERSION},
            },
        }
    if method == "ping":
        return {"jsonrpc": "2.0", "id": request_id, "result": {}}

Copyif method == "initialize":
    requested = (message.get("params") or {}).get("protocolVersion")
    return {
        "jsonrpc": "2.0",
        "id": request_id,
        "result": {
            "protocolVersion": requested or PROTOCOL_VERSION,
            "capabilities": {"tools": {"listChanged": False}},
            "serverInfo": {"name": "mem0", "version": PLUGIN_VERSION},
        },
    }
if method == "ping":
    return {"jsonrpc": "2.0", "id": request_id, "result": {}}
if method == "tools/list":
    return {
        "jsonrpc": "2.0",
        "id": request_id,
        "result": {
            "tools": [
                {
                    "name": TOOL_NAME,
                    "description": TOOL_DESCRIPTION,
                    "inputSchema": TOOL_SCHEMA,
                    "annotations": {
                        "readOnlyHint": True,
                        "idempotentHint": True,
                        "openWorldHint": True,
                    },
                }
            ]
        },
    }
…
if method == "tools/call":
    params = message.get("params") or {}
    if params.get("name") != TOOL_NAME:
        result = _tool_response("Unknown Mem0 tool.", is_error=True)
    else:
        try:
            result = _tool_response(
                call_search_memories(params.get("arguments"), _workspace_cwd(params))
            )
        except ToolInputError as exc:
            result = _tool_response(str(exc), is_error=True)
        except Exception:
            result = _tool_response("Memory search failed.", is_error=True)
    return {"jsonrpc": "2.0", "id": request_id, "result": result}

Copyparams = message.get("params") or {}
if params.get("name") != TOOL_NAME:
    result = _tool_response("Unknown Mem0 tool.", is_error=True)
else:
    try:
        result = _tool_response(
            call_search_memories(params.get("arguments"), _workspace_cwd(params))
        )
    except ToolInputError as exc:
        result = _tool_response(str(exc), is_error=True)
    except Exception:
        result = _tool_response("Memory search failed.", is_error=True)
return {"jsonrpc": "2.0", "id": request_id, "result": result}

CopyTOOL_SCHEMA = {
    "type": "object",
    "properties": {
        "query": {
            "type": "string",
            "minLength": 1,
            "maxLength": 2000,
            "description": "A direct question about earlier work in this repository.",
        },
        "top_k": {
            "type": "integer",
            "minimum": 1,
            "maximum": 20,
            "description": "Maximum memories to return. Uses Mem0's configured default when omitted.",
        },
        "category": {
            "type": "string",
            "enum": list(CODING_MEMORY_CATEGORY_NAMES),
            "description": "Optional memory category. Omit to search every category.",
        },
        "scope": {
            "type": "string",
            "enum": list(SEARCH_SCOPES),
            "description": (
                "Which memories to search. 'repo' (default) is the whole repository's "
                "shared memory plus your own preferences, 'dir' narrows the shared "
                "part to the current directory, 'mine' is your preferences alone."
            ),
        },
        "run_id": {
            "type": "string",
            "minLength": 1,
            "description": (
                "Optional coding-agent session ID. With any scope, restricts results to memories "
                "saved in that session. Omit to recall memories across sessions."
            ),
        },
    },
    "required": ["query"],
    "additionalProperties": False,
}
…
def _validate_arguments(
    arguments: Any,
) -> tuple[str, int | None, str | None, str | None, str | None]:
    if not isinstance(arguments, dict):
        raise ToolInputError("Search arguments must be an object.")
 
    unknown = set(arguments) - {"query", "top_k", "category", "scope", "run_id"}
    if unknown:
        raise ToolInputError(f"Unknown search argument: {sorted(unknown)[0]}")
 
    query = arguments.get("query")
    if not isinstance(query, str) or not query.strip():
        raise ToolInputError("query must be a non-empty string.")
    query = query.strip()
    if len(query) > 2000:
        raise ToolInputError("query must be at most 2,000 characters.")
 
    top_k = arguments.get("top_k")
    if top_k is not None and (
        isinstance(top_k, bool) or not isinstance(top_k, int) or not 1 <= top_k <= 20
    ):
        raise ToolInputError("top_k must be an integer from 1 to 20.")
 
    category = arguments.get("category")
    if category is not None and category not in CODING_MEMORY_CATEGORY_NAMES:
        raise ToolInputError("category must be one of Mem0's supported categories.")
 
    scope = arguments.get("scope")
    if scope is not None and scope not in SEARCH_SCOPES:
        raise ToolInputError(f"scope must be one of {list(SEARCH_SCOPES)}.")
 
    run_id = arguments.get("run_id")
    if run_id is not None:
        if not isinstance(run_id, str) or not run_id.strip():
            raise ToolInputError("run_id must be a non-empty string.")
        run_id = run_id.strip()
 
    return query, top_k, category, scope, run_id

Copy"scope": {
    "type": "string",
    "enum": list(SEARCH_SCOPES),
    "description": (
        "Which memories to search. 'repo' (default) is the whole repository's "
        "shared memory plus your own preferences, 'dir' narrows the shared "
        "part to the current directory, 'mine' is your preferences alone."
    ),
},
…
category = arguments.get("category")
if category is not None and category not in CODING_MEMORY_CATEGORY_NAMES:
    raise ToolInputError("category must be one of Mem0's supported categories.")
 
scope = arguments.get("scope")
if scope is not None and scope not in SEARCH_SCOPES:
    raise ToolInputError(f"scope must be one of {list(SEARCH_SCOPES)}.")