Scott.Idem/Cortex-Inara
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: improve AE Journal read toolset

Browse Source 
- ae_journal_entry_read: expose full entry content by id_random (title,
  journal, tags, summary, full content with configurable truncation)
- ae_journal_entries_list: browse all entries in a journal newest-first,
  numbered with id/title/tags/summary/date and pagination support
- ae_journal_search: richer output — tags, updated date, 400-char preview
  (was 200), show summary OR preview (not both when summary exists)

_get_entry() was already implemented; read tool just exposes it properly.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Scott Idem
2026-04-30 19:47:59 -04:00
parent 36fdda6728
commit 77327d97ad
2 changed files with 188 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

57
cortex/tools/__init__.py
View File

@@ -18,6 +18,8 @@ from google.genai import types
from tools.web import search as _web_search
from tools.ae_knowledge import journal_search as _ae_journal_search
from tools.ae_knowledge import journal_list as _ae_journal_list
from tools.ae_knowledge import journal_entry_read as _ae_journal_entry_read
from tools.ae_knowledge import journal_entries_list as _ae_journal_entries_list
from tools.ae_knowledge import journal_entry_create as _ae_journal_entry_create
from tools.ae_knowledge import journal_entry_update as _ae_journal_entry_update
from tools.ae_knowledge import journal_entry_disable as _ae_journal_entry_disable
@@ -119,6 +121,57 @@ _ae_journal_search_declaration = types.FunctionDeclaration(
),
)
_ae_journal_entry_read_declaration = types.FunctionDeclaration(
name="ae_journal_entry_read",
description=(
"Fetch the full content of a single journal entry by its id_random. "
"Use this when you need to read an entry before editing it, or when search results "
"don't show enough content. Returns title, journal, tags, summary, and full content."
),
parameters=types.Schema(
type=types.Type.OBJECT,
properties={
"entry_id": types.Schema(
type=types.Type.STRING,
description="The id_random of the journal entry to read.",
),
"max_content_chars": types.Schema(
type=types.Type.INTEGER,
description="Maximum characters of content to return (default 4000). Increase for long entries.",
),
},
required=["entry_id"],
),
)
_ae_journal_entries_list_declaration = types.FunctionDeclaration(
name="ae_journal_entries_list",
description=(
"List entries in a specific journal, newest first. "
"Use this to browse what's in a journal when you don't have a search keyword, "
"or to find entries by browsing rather than searching. "
"Returns numbered entries with id, title, tags, summary, and date."
),
parameters=types.Schema(
type=types.Type.OBJECT,
properties={
"journal_id": types.Schema(
type=types.Type.STRING,
description="The id_random of the journal to list entries from.",
),
"max_results": types.Schema(
type=types.Type.INTEGER,
description="Number of entries to return (default 20, max 50).",
),
"page": types.Schema(
type=types.Type.INTEGER,
description="Page number for pagination (default 1).",
),
},
required=["journal_id"],
),
)
_ae_journal_entry_create_declaration = types.FunctionDeclaration(
name="ae_journal_entry_create",
description=(
@@ -283,6 +336,8 @@ _CALLABLES: dict[str, callable] = {
"web_search": _web_search,
"ae_journal_list": _ae_journal_list,
"ae_journal_search": _ae_journal_search,
"ae_journal_entry_read": _ae_journal_entry_read,
"ae_journal_entries_list": _ae_journal_entries_list,
"ae_journal_entry_create": _ae_journal_entry_create,
"ae_journal_entry_update": _ae_journal_entry_update,
"ae_journal_entry_disable": _ae_journal_entry_disable,
@@ -858,6 +913,8 @@ _ALL_DECLARATIONS: list[types.FunctionDeclaration] = [
_web_search_declaration,
_ae_journal_list_declaration,
_ae_journal_search_declaration,
_ae_journal_entry_read_declaration,
_ae_journal_entries_list_declaration,
_ae_journal_entry_create_declaration,
_ae_journal_entry_update_declaration,
_ae_journal_entry_disable_declaration,

139
cortex/tools/ae_knowledge.py
View File

@@ -91,23 +91,27 @@ def _sync_journal_search(query: str, journal_id: str | None, max_results: int) -
lines = [f"Journal entries matching **{query}** ({len(entries)} result(s)):\n"]
for entry in entries:
title = entry.get("name") or "(untitled)"
title = entry.get("name") or "(untitled)"
entry_id = entry.get("id_random", "")
journal_name = entry.get("journal_name") or entry.get("parent_name") or ""
summary = entry.get("summary") or ""
content_preview = (entry.get("content") or "")[:200].replace("\n", " ")
summary = entry.get("summary") or ""
tags = entry.get("tags") or []
updated = (entry.get("updated_at") or entry.get("created_at") or "")[:10]
content_preview = (entry.get("content") or "")[:400].replace("\n", " ")
header = f"**{title}**"
if journal_name:
header += f" ({journal_name})"
if entry_id:
header += f" — id: `{entry_id}`"
header += f" — id: `{entry_id}`"
if updated:
header += f" [{updated}]"
lines.append(header)
if tags:
lines.append(f" Tags: {', '.join(tags)}")
if summary:
lines.append(f" Summary: {summary}")
if content_preview:
lines.append(f" {content_preview}…")
elif content_preview:
lines.append(f" {content_preview}{'…' if len(entry.get('content','')) > 400 else ''}")
lines.append("")
return "\n".join(lines).strip()
@@ -264,6 +268,125 @@ def _patch_entry(entry_id: str, payload: dict) -> str:
return f"Error updating entry {entry_id}: {e}"
# ---------------------------------------------------------------------------
# Tool: ae_journal_entry_read
# ---------------------------------------------------------------------------
async def journal_entry_read(entry_id: str, max_content_chars: int = 4000) -> str:
"""Return the full content of a single journal entry by its id_random."""
err = _check_config()
if err:
return err
return await asyncio.to_thread(_sync_journal_entry_read, entry_id, max_content_chars)
def _sync_journal_entry_read(entry_id: str, max_content_chars: int) -> str:
entry = _get_entry(entry_id)
if isinstance(entry, str):
return entry
title = entry.get("name") or "(untitled)"
journal = entry.get("journal_name") or entry.get("parent_name") or ""
summary = entry.get("summary") or ""
tags = entry.get("tags") or []
content = entry.get("content") or ""
updated = (entry.get("updated_at") or entry.get("created_at") or "")[:19].replace("T", " ")
enabled = entry.get("enable", True)
lines = [f"# {title}"]
meta: list[str] = [f"id: `{entry_id}`"]
if journal:
meta.append(f"journal: {journal}")
if updated:
meta.append(f"updated: {updated}")
if not enabled:
meta.append("**DISABLED**")
lines.append(" ".join(meta))
if tags:
lines.append(f"Tags: {', '.join(tags)}")
if summary:
lines.append(f"\nSummary: {summary}")
lines.append("\n---\n")
truncated = len(content) > max_content_chars
lines.append(content[:max_content_chars])
if truncated:
lines.append(
f"\n\n[Content truncated at {max_content_chars} chars — "
f"{len(content)} total. Call again with a higher max_content_chars to read more.]"
)
return "\n".join(lines)
# ---------------------------------------------------------------------------
# Tool: ae_journal_entries_list
# ---------------------------------------------------------------------------
async def journal_entries_list(journal_id: str, max_results: int = 20, page: int = 1) -> str:
"""List entries in a specific journal, newest first."""
err = _check_config()
if err:
return err
return await asyncio.to_thread(_sync_journal_entries_list, journal_id, max_results, page)
def _sync_journal_entries_list(journal_id: str, max_results: int, page: int) -> str:
import requests
url = f"{settings.ae_api_url}/v3/crud/journal_entry/search"
search_body: dict = {
"page_size": max_results,
"page": page,
"order_by": "-updated_at",
}
params = {"for_obj_type": "journal", "for_obj_id": journal_id}
try:
resp = requests.post(
url,
headers=_headers(),
params=params,
json=search_body,
timeout=settings.ae_api_timeout,
)
resp.raise_for_status()
data = resp.json()
except Exception as e:
logger.warning("ae_journal_entries_list failed: %s", e)
return f"Journal entries list error: {e}"
entries = data.get("data", [])
total = data.get("total") or data.get("count") or len(entries)
if not entries:
return f"No entries found in journal `{journal_id}`."
offset = (page - 1) * max_results + 1
lines = [f"Entries in journal `{journal_id}` — showing {offset}–{offset + len(entries) - 1} of {total}:\n"]
for i, entry in enumerate(entries, offset):
title = entry.get("name") or "(untitled)"
entry_id = entry.get("id_random", "")
tags = entry.get("tags") or []
summary = entry.get("summary") or ""
updated = (entry.get("updated_at") or entry.get("created_at") or "")[:10]
enabled = entry.get("enable", True)
status = "" if enabled else " [disabled]"
date_str = f" [{updated}]" if updated else ""
lines.append(f"{i}. **{title}**{status} — id: `{entry_id}`{date_str}")
if tags:
lines.append(f" Tags: {', '.join(tags)}")
if summary:
lines.append(f" {summary[:150]}{'…' if len(summary) > 150 else ''}")
lines.append("")
if total > offset + len(entries) - 1:
lines.append(f"(More entries available — call again with page={page + 1})")
return "\n".join(lines).rstrip()
# ---------------------------------------------------------------------------
# Tool: ae_journal_entry_update
# ---------------------------------------------------------------------------