mirror of
https://github.com/usestrix/strix.git
synced 2026-08-23 03:12:37 +02:00
refactor: inline non-sandbox actions, strip registry, drop schemas
Cleanup pass after the migration: #1 Inline ``*_actions.py`` into wrapper ``tool[s].py`` for the non-sandbox tools (think, todo, notes, reporting, web_search, finish_scan). One file per tool family now. Helpers + public function bodies live alongside the ``@strix_tool``-decorated wrappers that call them. For notes, the sync helpers are renamed to ``_create_note_impl`` / ``_list_notes_impl`` / etc. so the public names ``create_note`` / ``list_notes`` / etc. can be the FunctionTool instances the agent factory imports. ``append_note_content`` (used by the agents-graph wiki-update hook) calls the impl helpers directly. #2 Delete ``strix/tools/_state_adapter.py``. The ``AgentStateAdapter`` shim only existed to feed legacy ``*_actions.py`` functions a ``state.agent_id`` they could read. With the actions inlined, the wrappers read ``ctx.context['agent_id']`` directly. #3 Strip ``strix/tools/registry.py`` from ~250 LOC to ~110. Deleted: XML schema loading, ``_parse_param_schema``, ``get_tools_prompt``, ``get_tool_param_schema``, ``needs_agent_state``, ``should_execute_in_sandbox``, ``validate_tool_availability`` — all for the host-side legacy dispatcher path. Kept the ``register_tool`` decorator (sandbox side), ``get_tool_by_name``, ``get_tool_names``, ``tools`` list, ``clear_registry``. The Jinja prompt template's ``{{ get_tools_prompt() }}`` injection is dropped — the SDK auto-generates tool descriptions from function signatures, so the legacy XML tool block was redundant and stale. #4 Delete every ``*_actions_schema.xml`` (12 files). They were read by the now-removed ``_load_xml_schema`` to build the legacy prompt's tool descriptions. No consumer remains. Side fixes: - ``reporting_renderer.py`` updated to import ``_parse_*_xml`` from the new location with leading underscore. - ``test_local_tools.py``, ``test_notes_jsonl_concurrency.py``, ``test_notes_wiki.py`` updated to point at the new module paths and call the ``_*_impl`` sync helpers. Tests: 279/279 passing. ~1500 LOC of action files moved into the tool wrappers; ~140 LOC of registry boilerplate removed; ~400 lines of dead XML deleted.
This commit is contained in:
0xallam
@@ -1,4 +1,5 @@
|
||||
from .notes_actions import (
|
||||
from .tools import (
|
||||
append_note_content,
|
||||
create_note,
|
||||
delete_note,
|
||||
get_note,
|
||||
@@ -8,6 +9,7 @@ from .notes_actions import (
|
||||
|
||||
|
||||
__all__ = [
|
||||
"append_note_content",
|
||||
"create_note",
|
||||
"delete_note",
|
||||
"get_note",
|
||||
|
||||
@@ -1,464 +0,0 @@
|
||||
import json
|
||||
import threading
|
||||
import uuid
|
||||
from datetime import UTC, datetime
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from strix.tools.registry import register_tool
|
||||
|
||||
|
||||
_notes_storage: dict[str, dict[str, Any]] = {}
|
||||
_VALID_NOTE_CATEGORIES = ["general", "findings", "methodology", "questions", "plan", "wiki"]
|
||||
_notes_lock = threading.RLock()
|
||||
_loaded_notes_run_dir: str | None = None
|
||||
_DEFAULT_CONTENT_PREVIEW_CHARS = 280
|
||||
|
||||
|
||||
def _get_run_dir() -> Path | None:
|
||||
try:
|
||||
from strix.telemetry.tracer import get_global_tracer
|
||||
|
||||
tracer = get_global_tracer()
|
||||
if not tracer:
|
||||
return None
|
||||
return tracer.get_run_dir()
|
||||
except (ImportError, OSError, RuntimeError):
|
||||
return None
|
||||
|
||||
|
||||
def _get_notes_jsonl_path() -> Path | None:
|
||||
run_dir = _get_run_dir()
|
||||
if not run_dir:
|
||||
return None
|
||||
|
||||
notes_dir = run_dir / "notes"
|
||||
notes_dir.mkdir(parents=True, exist_ok=True)
|
||||
return notes_dir / "notes.jsonl"
|
||||
|
||||
|
||||
def _append_note_event(op: str, note_id: str, note: dict[str, Any] | None = None) -> None:
|
||||
"""Append one note operation to the run's ``notes/notes.jsonl``.
|
||||
|
||||
C6 (AUDIT_R2.md §1.1): hold ``_notes_lock`` across the file open + write
|
||||
so two concurrent agents (or two parallel SDK tool calls in Phase 6)
|
||||
cannot interleave bytes mid-line and corrupt the JSONL.
|
||||
"""
|
||||
notes_path = _get_notes_jsonl_path()
|
||||
if not notes_path:
|
||||
return
|
||||
|
||||
event: dict[str, Any] = {
|
||||
"timestamp": datetime.now(UTC).isoformat(),
|
||||
"op": op,
|
||||
"note_id": note_id,
|
||||
}
|
||||
if note is not None:
|
||||
event["note"] = note
|
||||
|
||||
with _notes_lock, notes_path.open("a", encoding="utf-8") as f:
|
||||
f.write(f"{json.dumps(event, ensure_ascii=True)}\n")
|
||||
|
||||
|
||||
def _load_notes_from_jsonl(notes_path: Path) -> dict[str, dict[str, Any]]:
|
||||
hydrated: dict[str, dict[str, Any]] = {}
|
||||
if not notes_path.exists():
|
||||
return hydrated
|
||||
|
||||
with notes_path.open(encoding="utf-8") as f:
|
||||
for raw_line in f:
|
||||
line = raw_line.strip()
|
||||
if not line:
|
||||
continue
|
||||
|
||||
try:
|
||||
event = json.loads(line)
|
||||
except json.JSONDecodeError:
|
||||
continue
|
||||
|
||||
op = str(event.get("op", "")).strip().lower()
|
||||
note_id = str(event.get("note_id", "")).strip()
|
||||
if not note_id or op not in {"create", "update", "delete"}:
|
||||
continue
|
||||
|
||||
if op == "delete":
|
||||
hydrated.pop(note_id, None)
|
||||
continue
|
||||
|
||||
note = event.get("note")
|
||||
if not isinstance(note, dict):
|
||||
continue
|
||||
|
||||
existing = hydrated.get(note_id, {})
|
||||
existing.update(note)
|
||||
hydrated[note_id] = existing
|
||||
|
||||
return hydrated
|
||||
|
||||
|
||||
def _ensure_notes_loaded() -> None:
|
||||
global _loaded_notes_run_dir # noqa: PLW0603
|
||||
|
||||
run_dir = _get_run_dir()
|
||||
run_dir_key = str(run_dir.resolve()) if run_dir else "__no_run_dir__"
|
||||
if _loaded_notes_run_dir == run_dir_key:
|
||||
return
|
||||
|
||||
_notes_storage.clear()
|
||||
|
||||
notes_path = _get_notes_jsonl_path()
|
||||
if notes_path:
|
||||
_notes_storage.update(_load_notes_from_jsonl(notes_path))
|
||||
try:
|
||||
for note_id, note in _notes_storage.items():
|
||||
if note.get("category") == "wiki":
|
||||
_persist_wiki_note(note_id, note)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
_loaded_notes_run_dir = run_dir_key
|
||||
|
||||
|
||||
def _sanitize_wiki_title(title: str) -> str:
|
||||
cleaned = "".join(ch.lower() if ch.isalnum() else "-" for ch in title.strip())
|
||||
slug = "-".join(part for part in cleaned.split("-") if part)
|
||||
return slug or "wiki-note"
|
||||
|
||||
|
||||
def _get_wiki_directory() -> Path | None:
|
||||
try:
|
||||
run_dir = _get_run_dir()
|
||||
if not run_dir:
|
||||
return None
|
||||
|
||||
wiki_dir = run_dir / "wiki"
|
||||
wiki_dir.mkdir(parents=True, exist_ok=True)
|
||||
except OSError:
|
||||
return None
|
||||
else:
|
||||
return wiki_dir
|
||||
|
||||
|
||||
def _get_wiki_note_path(note_id: str, note: dict[str, Any]) -> Path | None:
|
||||
wiki_dir = _get_wiki_directory()
|
||||
if not wiki_dir:
|
||||
return None
|
||||
|
||||
wiki_filename = note.get("wiki_filename")
|
||||
if not isinstance(wiki_filename, str) or not wiki_filename.strip():
|
||||
title = note.get("title", "wiki-note")
|
||||
wiki_filename = f"{note_id}-{_sanitize_wiki_title(str(title))}.md"
|
||||
note["wiki_filename"] = wiki_filename
|
||||
|
||||
return wiki_dir / wiki_filename
|
||||
|
||||
|
||||
def _persist_wiki_note(note_id: str, note: dict[str, Any]) -> None:
|
||||
wiki_path = _get_wiki_note_path(note_id, note)
|
||||
if not wiki_path:
|
||||
return
|
||||
|
||||
tags = note.get("tags", [])
|
||||
tags_line = ", ".join(str(tag) for tag in tags) if isinstance(tags, list) and tags else "none"
|
||||
|
||||
content = (
|
||||
f"# {note.get('title', 'Wiki Note')}\n\n"
|
||||
f"**Note ID:** {note_id}\n"
|
||||
f"**Created:** {note.get('created_at', '')}\n"
|
||||
f"**Updated:** {note.get('updated_at', '')}\n"
|
||||
f"**Tags:** {tags_line}\n\n"
|
||||
"## Content\n\n"
|
||||
f"{note.get('content', '')}\n"
|
||||
)
|
||||
wiki_path.write_text(content, encoding="utf-8")
|
||||
|
||||
|
||||
def _remove_wiki_note(note_id: str, note: dict[str, Any]) -> None:
|
||||
wiki_path = _get_wiki_note_path(note_id, note)
|
||||
if not wiki_path:
|
||||
return
|
||||
|
||||
if wiki_path.exists():
|
||||
wiki_path.unlink()
|
||||
|
||||
|
||||
def _filter_notes(
|
||||
category: str | None = None,
|
||||
tags: list[str] | None = None,
|
||||
search_query: str | None = None,
|
||||
) -> list[dict[str, Any]]:
|
||||
_ensure_notes_loaded()
|
||||
filtered_notes = []
|
||||
|
||||
for note_id, note in _notes_storage.items():
|
||||
if category and note.get("category") != category:
|
||||
continue
|
||||
|
||||
if tags:
|
||||
note_tags = note.get("tags", [])
|
||||
if not any(tag in note_tags for tag in tags):
|
||||
continue
|
||||
|
||||
if search_query:
|
||||
search_lower = search_query.lower()
|
||||
title_match = search_lower in note.get("title", "").lower()
|
||||
content_match = search_lower in note.get("content", "").lower()
|
||||
if not (title_match or content_match):
|
||||
continue
|
||||
|
||||
note_with_id = note.copy()
|
||||
note_with_id["note_id"] = note_id
|
||||
filtered_notes.append(note_with_id)
|
||||
|
||||
filtered_notes.sort(key=lambda x: x.get("created_at", ""), reverse=True)
|
||||
return filtered_notes
|
||||
|
||||
|
||||
def _to_note_listing_entry(
|
||||
note: dict[str, Any],
|
||||
*,
|
||||
include_content: bool = False,
|
||||
) -> dict[str, Any]:
|
||||
entry = {
|
||||
"note_id": note.get("note_id"),
|
||||
"title": note.get("title", ""),
|
||||
"category": note.get("category", "general"),
|
||||
"tags": note.get("tags", []),
|
||||
"created_at": note.get("created_at", ""),
|
||||
"updated_at": note.get("updated_at", ""),
|
||||
}
|
||||
|
||||
wiki_filename = note.get("wiki_filename")
|
||||
if isinstance(wiki_filename, str) and wiki_filename:
|
||||
entry["wiki_filename"] = wiki_filename
|
||||
|
||||
content = str(note.get("content", ""))
|
||||
if include_content:
|
||||
entry["content"] = content
|
||||
elif content:
|
||||
if len(content) > _DEFAULT_CONTENT_PREVIEW_CHARS:
|
||||
entry["content_preview"] = f"{content[:_DEFAULT_CONTENT_PREVIEW_CHARS].rstrip()}..."
|
||||
else:
|
||||
entry["content_preview"] = content
|
||||
|
||||
return entry
|
||||
|
||||
|
||||
@register_tool(sandbox_execution=False)
|
||||
def create_note( # noqa: PLR0911
|
||||
title: str,
|
||||
content: str,
|
||||
category: str = "general",
|
||||
tags: list[str] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
with _notes_lock:
|
||||
try:
|
||||
_ensure_notes_loaded()
|
||||
|
||||
if not title or not title.strip():
|
||||
return {"success": False, "error": "Title cannot be empty", "note_id": None}
|
||||
|
||||
if not content or not content.strip():
|
||||
return {"success": False, "error": "Content cannot be empty", "note_id": None}
|
||||
|
||||
if category not in _VALID_NOTE_CATEGORIES:
|
||||
return {
|
||||
"success": False,
|
||||
"error": (
|
||||
f"Invalid category. Must be one of: {', '.join(_VALID_NOTE_CATEGORIES)}"
|
||||
),
|
||||
"note_id": None,
|
||||
}
|
||||
|
||||
note_id = ""
|
||||
for _ in range(20):
|
||||
candidate = str(uuid.uuid4())[:5]
|
||||
if candidate not in _notes_storage:
|
||||
note_id = candidate
|
||||
break
|
||||
if not note_id:
|
||||
return {"success": False, "error": "Failed to allocate note ID", "note_id": None}
|
||||
|
||||
timestamp = datetime.now(UTC).isoformat()
|
||||
|
||||
note = {
|
||||
"title": title.strip(),
|
||||
"content": content.strip(),
|
||||
"category": category,
|
||||
"tags": tags or [],
|
||||
"created_at": timestamp,
|
||||
"updated_at": timestamp,
|
||||
}
|
||||
|
||||
_notes_storage[note_id] = note
|
||||
_append_note_event("create", note_id, note)
|
||||
if category == "wiki":
|
||||
_persist_wiki_note(note_id, note)
|
||||
|
||||
except (ValueError, TypeError) as e:
|
||||
return {"success": False, "error": f"Failed to create note: {e}", "note_id": None}
|
||||
except OSError as e:
|
||||
return {"success": False, "error": f"Failed to persist wiki note: {e}", "note_id": None}
|
||||
else:
|
||||
return {
|
||||
"success": True,
|
||||
"note_id": note_id,
|
||||
"message": f"Note '{title}' created successfully",
|
||||
}
|
||||
|
||||
|
||||
@register_tool(sandbox_execution=False)
|
||||
def list_notes(
|
||||
category: str | None = None,
|
||||
tags: list[str] | None = None,
|
||||
search: str | None = None,
|
||||
include_content: bool = False,
|
||||
) -> dict[str, Any]:
|
||||
with _notes_lock:
|
||||
try:
|
||||
filtered_notes = _filter_notes(category=category, tags=tags, search_query=search)
|
||||
notes = [
|
||||
_to_note_listing_entry(note, include_content=include_content)
|
||||
for note in filtered_notes
|
||||
]
|
||||
|
||||
return {
|
||||
"success": True,
|
||||
"notes": notes,
|
||||
"total_count": len(notes),
|
||||
}
|
||||
|
||||
except (ValueError, TypeError) as e:
|
||||
return {
|
||||
"success": False,
|
||||
"error": f"Failed to list notes: {e}",
|
||||
"notes": [],
|
||||
"total_count": 0,
|
||||
}
|
||||
|
||||
|
||||
@register_tool(sandbox_execution=False)
|
||||
def get_note(note_id: str) -> dict[str, Any]:
|
||||
with _notes_lock:
|
||||
try:
|
||||
_ensure_notes_loaded()
|
||||
|
||||
if not note_id or not note_id.strip():
|
||||
return {
|
||||
"success": False,
|
||||
"error": "Note ID cannot be empty",
|
||||
"note": None,
|
||||
}
|
||||
|
||||
note = _notes_storage.get(note_id)
|
||||
if note is None:
|
||||
return {
|
||||
"success": False,
|
||||
"error": f"Note with ID '{note_id}' not found",
|
||||
"note": None,
|
||||
}
|
||||
|
||||
note_with_id = note.copy()
|
||||
note_with_id["note_id"] = note_id
|
||||
|
||||
except (ValueError, TypeError) as e:
|
||||
return {
|
||||
"success": False,
|
||||
"error": f"Failed to get note: {e}",
|
||||
"note": None,
|
||||
}
|
||||
else:
|
||||
return {"success": True, "note": note_with_id}
|
||||
|
||||
|
||||
def append_note_content(note_id: str, delta: str) -> dict[str, Any]:
|
||||
with _notes_lock:
|
||||
try:
|
||||
_ensure_notes_loaded()
|
||||
|
||||
if note_id not in _notes_storage:
|
||||
return {"success": False, "error": f"Note with ID '{note_id}' not found"}
|
||||
|
||||
note = _notes_storage[note_id]
|
||||
existing_content = str(note.get("content") or "")
|
||||
updated_content = f"{existing_content.rstrip()}{delta}"
|
||||
result: dict[str, Any] = update_note(
|
||||
note_id=note_id,
|
||||
content=updated_content,
|
||||
)
|
||||
except (ValueError, TypeError) as e:
|
||||
return {"success": False, "error": f"Failed to append note content: {e}"}
|
||||
else:
|
||||
return result
|
||||
|
||||
|
||||
@register_tool(sandbox_execution=False)
|
||||
def update_note(
|
||||
note_id: str,
|
||||
title: str | None = None,
|
||||
content: str | None = None,
|
||||
tags: list[str] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
with _notes_lock:
|
||||
try:
|
||||
_ensure_notes_loaded()
|
||||
|
||||
if note_id not in _notes_storage:
|
||||
return {"success": False, "error": f"Note with ID '{note_id}' not found"}
|
||||
|
||||
note = _notes_storage[note_id]
|
||||
|
||||
if title is not None:
|
||||
if not title.strip():
|
||||
return {"success": False, "error": "Title cannot be empty"}
|
||||
note["title"] = title.strip()
|
||||
|
||||
if content is not None:
|
||||
if not content.strip():
|
||||
return {"success": False, "error": "Content cannot be empty"}
|
||||
note["content"] = content.strip()
|
||||
|
||||
if tags is not None:
|
||||
note["tags"] = tags
|
||||
|
||||
note["updated_at"] = datetime.now(UTC).isoformat()
|
||||
_append_note_event("update", note_id, note)
|
||||
if note.get("category") == "wiki":
|
||||
_persist_wiki_note(note_id, note)
|
||||
|
||||
return {
|
||||
"success": True,
|
||||
"message": f"Note '{note['title']}' updated successfully",
|
||||
}
|
||||
|
||||
except (ValueError, TypeError) as e:
|
||||
return {"success": False, "error": f"Failed to update note: {e}"}
|
||||
except OSError as e:
|
||||
return {"success": False, "error": f"Failed to persist wiki note: {e}"}
|
||||
|
||||
|
||||
@register_tool(sandbox_execution=False)
|
||||
def delete_note(note_id: str) -> dict[str, Any]:
|
||||
with _notes_lock:
|
||||
try:
|
||||
_ensure_notes_loaded()
|
||||
|
||||
if note_id not in _notes_storage:
|
||||
return {"success": False, "error": f"Note with ID '{note_id}' not found"}
|
||||
|
||||
note = _notes_storage[note_id]
|
||||
note_title = note["title"]
|
||||
if note.get("category") == "wiki":
|
||||
_remove_wiki_note(note_id, note)
|
||||
del _notes_storage[note_id]
|
||||
_append_note_event("delete", note_id)
|
||||
|
||||
except (ValueError, TypeError) as e:
|
||||
return {"success": False, "error": f"Failed to delete note: {e}"}
|
||||
except OSError as e:
|
||||
return {"success": False, "error": f"Failed to delete wiki note: {e}"}
|
||||
else:
|
||||
return {
|
||||
"success": True,
|
||||
"message": f"Note '{note_title}' deleted successfully",
|
||||
}
|
||||
@@ -1,180 +0,0 @@
|
||||
<tools>
|
||||
<tool name="create_note">
|
||||
<description>Create a personal note for observations, findings, and research during the scan.</description>
|
||||
<details>Use this tool for documenting discoveries, observations, methodology notes, and questions.
|
||||
This is your personal and shared run memory for recording information you want to remember or reference later.
|
||||
Use category "wiki" for repository source maps shared across agents in the same run.
|
||||
For tracking actionable tasks, use the todo tool instead.</details>
|
||||
<parameters>
|
||||
<parameter name="title" type="string" required="true">
|
||||
<description>Title of the note</description>
|
||||
</parameter>
|
||||
<parameter name="content" type="string" required="true">
|
||||
<description>Content of the note</description>
|
||||
</parameter>
|
||||
<parameter name="category" type="string" required="false">
|
||||
<description>Category to organize the note (default: "general", "findings", "methodology", "questions", "plan", "wiki")</description>
|
||||
</parameter>
|
||||
<parameter name="tags" type="string" required="false">
|
||||
<description>Tags for categorization</description>
|
||||
</parameter>
|
||||
</parameters>
|
||||
<returns type="Dict[str, Any]">
|
||||
<description>Response containing: - note_id: ID of the created note - success: Whether the note was created successfully</description>
|
||||
</returns>
|
||||
<examples>
|
||||
# Document an interesting finding
|
||||
<function=create_note>
|
||||
<parameter=title>Authentication Bypass Findings</parameter>
|
||||
<parameter=content>Discovered multiple authentication bypass vectors in the login system:
|
||||
|
||||
1. SQL Injection in username field
|
||||
- Payload: admin'--
|
||||
- Result: Full authentication bypass
|
||||
- Endpoint: POST /api/v1/auth/login
|
||||
|
||||
2. JWT Token Weakness
|
||||
- Algorithm confusion attack possible (RS256 -> HS256)
|
||||
- Token expiration is 24 hours but no refresh rotation
|
||||
- Token stored in localStorage (XSS risk)
|
||||
|
||||
3. Password Reset Flow
|
||||
- Reset tokens are only 6 digits (brute-forceable)
|
||||
- No rate limiting on reset attempts
|
||||
- Token valid for 48 hours
|
||||
|
||||
Next Steps:
|
||||
- Extract full database via SQL injection
|
||||
- Test JWT manipulation attacks
|
||||
- Attempt password reset brute force</parameter>
|
||||
<parameter=category>findings</parameter>
|
||||
<parameter=tags>["auth", "sqli", "jwt", "critical"]</parameter>
|
||||
</function>
|
||||
|
||||
# Methodology note
|
||||
<function=create_note>
|
||||
<parameter=title>API Endpoint Mapping Complete</parameter>
|
||||
<parameter=content>Completed comprehensive API enumeration using multiple techniques:
|
||||
|
||||
Discovered Endpoints:
|
||||
- /api/v1/auth/* - Authentication endpoints (login, register, reset)
|
||||
- /api/v1/users/* - User management (profile, settings, admin)
|
||||
- /api/v1/orders/* - Order management (IDOR vulnerability confirmed)
|
||||
- /api/v1/admin/* - Admin panel (403 but may be bypassable)
|
||||
- /api/internal/* - Internal APIs (should not be exposed)
|
||||
|
||||
Methods Used:
|
||||
- Analyzed JavaScript bundles for API calls
|
||||
- Bruteforced common paths with ffuf
|
||||
- Reviewed OpenAPI/Swagger documentation at /api/docs
|
||||
- Monitored traffic during normal application usage
|
||||
|
||||
Priority Targets:
|
||||
The /api/internal/* endpoints are high priority as they appear to lack authentication checks based on error message differences.</parameter>
|
||||
<parameter=category>methodology</parameter>
|
||||
<parameter=tags>["api", "enumeration", "recon"]</parameter>
|
||||
</function>
|
||||
</examples>
|
||||
</tool>
|
||||
<tool name="delete_note">
|
||||
<description>Delete a note.</description>
|
||||
<parameters>
|
||||
<parameter name="note_id" type="string" required="true">
|
||||
<description>ID of the note to delete</description>
|
||||
</parameter>
|
||||
</parameters>
|
||||
<returns type="Dict[str, Any]">
|
||||
<description>Response containing: - success: Whether the note was deleted successfully</description>
|
||||
</returns>
|
||||
<examples>
|
||||
<function=delete_note>
|
||||
<parameter=note_id>note_123</parameter>
|
||||
</function>
|
||||
</examples>
|
||||
</tool>
|
||||
<tool name="list_notes">
|
||||
<description>List existing notes with optional filtering and search (metadata-first by default).</description>
|
||||
<parameters>
|
||||
<parameter name="category" type="string" required="false">
|
||||
<description>Filter by category</description>
|
||||
</parameter>
|
||||
<parameter name="tags" type="string" required="false">
|
||||
<description>Filter by tags (returns notes with any of these tags)</description>
|
||||
</parameter>
|
||||
<parameter name="search" type="string" required="false">
|
||||
<description>Search query to find in note titles and content</description>
|
||||
</parameter>
|
||||
<parameter name="include_content" type="boolean" required="false">
|
||||
<description>Include full note content in each list item (default: false)</description>
|
||||
</parameter>
|
||||
</parameters>
|
||||
<returns type="Dict[str, Any]">
|
||||
<description>Response containing: - notes: List of matching notes (metadata + optional content/content_preview) - total_count: Total number of notes found</description>
|
||||
</returns>
|
||||
<examples>
|
||||
# List all findings
|
||||
<function=list_notes>
|
||||
<parameter=category>findings</parameter>
|
||||
</function>
|
||||
|
||||
# Search for SQL injection related notes
|
||||
<function=list_notes>
|
||||
<parameter=search>SQL injection</parameter>
|
||||
</function>
|
||||
|
||||
# Search within a specific category
|
||||
<function=list_notes>
|
||||
<parameter=search>admin</parameter>
|
||||
<parameter=category>findings</parameter>
|
||||
</function>
|
||||
|
||||
# Load shared repository wiki notes
|
||||
<function=list_notes>
|
||||
<parameter=category>wiki</parameter>
|
||||
</function>
|
||||
</examples>
|
||||
</tool>
|
||||
<tool name="get_note">
|
||||
<description>Get a single note by ID, including full content.</description>
|
||||
<parameters>
|
||||
<parameter name="note_id" type="string" required="true">
|
||||
<description>ID of the note to fetch</description>
|
||||
</parameter>
|
||||
</parameters>
|
||||
<returns type="Dict[str, Any]">
|
||||
<description>Response containing: - note: Note object including content - success: Whether note lookup succeeded</description>
|
||||
</returns>
|
||||
<examples>
|
||||
# Read a specific wiki note after listing note IDs
|
||||
<function=get_note>
|
||||
<parameter=note_id>abc12</parameter>
|
||||
</function>
|
||||
</examples>
|
||||
</tool>
|
||||
<tool name="update_note">
|
||||
<description>Update an existing note.</description>
|
||||
<parameters>
|
||||
<parameter name="note_id" type="string" required="true">
|
||||
<description>ID of the note to update</description>
|
||||
</parameter>
|
||||
<parameter name="title" type="string" required="false">
|
||||
<description>New title for the note</description>
|
||||
</parameter>
|
||||
<parameter name="content" type="string" required="false">
|
||||
<description>New content for the note</description>
|
||||
</parameter>
|
||||
<parameter name="tags" type="string" required="false">
|
||||
<description>New tags for the note</description>
|
||||
</parameter>
|
||||
</parameters>
|
||||
<returns type="Dict[str, Any]">
|
||||
<description>Response containing: - success: Whether the note was updated successfully</description>
|
||||
</returns>
|
||||
<examples>
|
||||
<function=update_note>
|
||||
<parameter=note_id>note_123</parameter>
|
||||
<parameter=content>Updated content with new findings...</parameter>
|
||||
</function>
|
||||
</examples>
|
||||
</tool>
|
||||
</tools>
|
||||
+431
-55
@@ -1,29 +1,422 @@
|
||||
"""SDK function-tool wrappers for the legacy notes tools.
|
||||
"""Per-run notes (shared across agents).
|
||||
|
||||
Five tools, all module-global (no per-agent silo). The legacy
|
||||
``notes_actions.py`` module already implements JSONL persistence and
|
||||
wiki Markdown rendering; these wrappers are pure delegation.
|
||||
|
||||
The C6 fix (lock-protected JSONL writes) was applied directly to the
|
||||
legacy module, so both code paths benefit.
|
||||
Persisted to ``run_dir/notes/notes.jsonl`` (replayable event log) and,
|
||||
for the ``wiki`` category, also rendered as Markdown to
|
||||
``run_dir/wiki/<slug>.md``. Concurrent appends are serialised by a
|
||||
threading.RLock so two agents writing simultaneously can't corrupt
|
||||
the JSONL.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
from typing import Any
|
||||
import logging
|
||||
import threading
|
||||
import uuid
|
||||
from datetime import UTC, datetime
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from pathlib import Path
|
||||
|
||||
from agents import RunContextWrapper
|
||||
|
||||
from strix.tools._decorator import strix_tool
|
||||
from strix.tools.notes import notes_actions as _impl
|
||||
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
_notes_storage: dict[str, dict[str, Any]] = {}
|
||||
_VALID_NOTE_CATEGORIES = ["general", "findings", "methodology", "questions", "plan", "wiki"]
|
||||
_notes_lock = threading.RLock()
|
||||
_loaded_notes_run_dir: str | None = None
|
||||
_DEFAULT_CONTENT_PREVIEW_CHARS = 280
|
||||
|
||||
|
||||
def _dump(result: dict[str, Any]) -> str:
|
||||
return json.dumps(result, ensure_ascii=False, default=str)
|
||||
|
||||
|
||||
def _get_run_dir() -> Path | None:
|
||||
try:
|
||||
from strix.telemetry.tracer import get_global_tracer
|
||||
|
||||
tracer = get_global_tracer()
|
||||
if not tracer:
|
||||
return None
|
||||
return tracer.get_run_dir()
|
||||
except (ImportError, OSError, RuntimeError):
|
||||
return None
|
||||
|
||||
|
||||
def _get_notes_jsonl_path() -> Path | None:
|
||||
run_dir = _get_run_dir()
|
||||
if not run_dir:
|
||||
return None
|
||||
notes_dir = run_dir / "notes"
|
||||
notes_dir.mkdir(parents=True, exist_ok=True)
|
||||
return notes_dir / "notes.jsonl"
|
||||
|
||||
|
||||
def _append_note_event(op: str, note_id: str, note: dict[str, Any] | None = None) -> None:
|
||||
"""Append one note operation to the run's ``notes/notes.jsonl``.
|
||||
|
||||
C6: hold ``_notes_lock`` across the file open + write so two
|
||||
concurrent agents can't interleave bytes mid-line.
|
||||
"""
|
||||
notes_path = _get_notes_jsonl_path()
|
||||
if not notes_path:
|
||||
return
|
||||
event: dict[str, Any] = {
|
||||
"timestamp": datetime.now(UTC).isoformat(),
|
||||
"op": op,
|
||||
"note_id": note_id,
|
||||
}
|
||||
if note is not None:
|
||||
event["note"] = note
|
||||
with _notes_lock, notes_path.open("a", encoding="utf-8") as f:
|
||||
f.write(f"{json.dumps(event, ensure_ascii=True)}\n")
|
||||
|
||||
|
||||
def _load_notes_from_jsonl(notes_path: Path) -> dict[str, dict[str, Any]]:
|
||||
hydrated: dict[str, dict[str, Any]] = {}
|
||||
if not notes_path.exists():
|
||||
return hydrated
|
||||
with notes_path.open(encoding="utf-8") as f:
|
||||
for raw_line in f:
|
||||
line = raw_line.strip()
|
||||
if not line:
|
||||
continue
|
||||
try:
|
||||
event = json.loads(line)
|
||||
except json.JSONDecodeError:
|
||||
continue
|
||||
op = str(event.get("op", "")).strip().lower()
|
||||
note_id = str(event.get("note_id", "")).strip()
|
||||
if not note_id or op not in {"create", "update", "delete"}:
|
||||
continue
|
||||
if op == "delete":
|
||||
hydrated.pop(note_id, None)
|
||||
continue
|
||||
note = event.get("note")
|
||||
if not isinstance(note, dict):
|
||||
continue
|
||||
existing = hydrated.get(note_id, {})
|
||||
existing.update(note)
|
||||
hydrated[note_id] = existing
|
||||
return hydrated
|
||||
|
||||
|
||||
def _ensure_notes_loaded() -> None:
|
||||
global _loaded_notes_run_dir # noqa: PLW0603
|
||||
run_dir = _get_run_dir()
|
||||
run_dir_key = str(run_dir.resolve()) if run_dir else "__no_run_dir__"
|
||||
if _loaded_notes_run_dir == run_dir_key:
|
||||
return
|
||||
_notes_storage.clear()
|
||||
notes_path = _get_notes_jsonl_path()
|
||||
if notes_path:
|
||||
_notes_storage.update(_load_notes_from_jsonl(notes_path))
|
||||
try:
|
||||
for note_id, note in _notes_storage.items():
|
||||
if note.get("category") == "wiki":
|
||||
_persist_wiki_note(note_id, note)
|
||||
except OSError:
|
||||
pass
|
||||
_loaded_notes_run_dir = run_dir_key
|
||||
|
||||
|
||||
def _sanitize_wiki_title(title: str) -> str:
|
||||
cleaned = "".join(ch.lower() if ch.isalnum() else "-" for ch in title.strip())
|
||||
slug = "-".join(part for part in cleaned.split("-") if part)
|
||||
return slug or "wiki-note"
|
||||
|
||||
|
||||
def _get_wiki_directory() -> Path | None:
|
||||
try:
|
||||
run_dir = _get_run_dir()
|
||||
if not run_dir:
|
||||
return None
|
||||
wiki_dir = run_dir / "wiki"
|
||||
wiki_dir.mkdir(parents=True, exist_ok=True)
|
||||
except OSError:
|
||||
return None
|
||||
else:
|
||||
return wiki_dir
|
||||
|
||||
|
||||
def _get_wiki_note_path(note_id: str, note: dict[str, Any]) -> Path | None:
|
||||
wiki_dir = _get_wiki_directory()
|
||||
if not wiki_dir:
|
||||
return None
|
||||
wiki_filename = note.get("wiki_filename")
|
||||
if not isinstance(wiki_filename, str) or not wiki_filename.strip():
|
||||
title = note.get("title", "wiki-note")
|
||||
wiki_filename = f"{note_id}-{_sanitize_wiki_title(str(title))}.md"
|
||||
note["wiki_filename"] = wiki_filename
|
||||
return wiki_dir / wiki_filename
|
||||
|
||||
|
||||
def _persist_wiki_note(note_id: str, note: dict[str, Any]) -> None:
|
||||
wiki_path = _get_wiki_note_path(note_id, note)
|
||||
if not wiki_path:
|
||||
return
|
||||
tags = note.get("tags", [])
|
||||
tags_line = ", ".join(str(tag) for tag in tags) if isinstance(tags, list) and tags else "none"
|
||||
content = (
|
||||
f"# {note.get('title', 'Wiki Note')}\n\n"
|
||||
f"**Note ID:** {note_id}\n"
|
||||
f"**Created:** {note.get('created_at', '')}\n"
|
||||
f"**Updated:** {note.get('updated_at', '')}\n"
|
||||
f"**Tags:** {tags_line}\n\n"
|
||||
"## Content\n\n"
|
||||
f"{note.get('content', '')}\n"
|
||||
)
|
||||
wiki_path.write_text(content, encoding="utf-8")
|
||||
|
||||
|
||||
def _remove_wiki_note(note_id: str, note: dict[str, Any]) -> None:
|
||||
wiki_path = _get_wiki_note_path(note_id, note)
|
||||
if not wiki_path:
|
||||
return
|
||||
if wiki_path.exists():
|
||||
wiki_path.unlink()
|
||||
|
||||
|
||||
def _filter_notes(
|
||||
category: str | None = None,
|
||||
tags: list[str] | None = None,
|
||||
search_query: str | None = None,
|
||||
) -> list[dict[str, Any]]:
|
||||
_ensure_notes_loaded()
|
||||
filtered: list[dict[str, Any]] = []
|
||||
for note_id, note in _notes_storage.items():
|
||||
if category and note.get("category") != category:
|
||||
continue
|
||||
if tags:
|
||||
note_tags = note.get("tags", [])
|
||||
if not any(tag in note_tags for tag in tags):
|
||||
continue
|
||||
if search_query:
|
||||
search_lower = search_query.lower()
|
||||
title_match = search_lower in note.get("title", "").lower()
|
||||
content_match = search_lower in note.get("content", "").lower()
|
||||
if not (title_match or content_match):
|
||||
continue
|
||||
entry = note.copy()
|
||||
entry["note_id"] = note_id
|
||||
filtered.append(entry)
|
||||
filtered.sort(key=lambda x: x.get("created_at", ""), reverse=True)
|
||||
return filtered
|
||||
|
||||
|
||||
def _to_note_listing_entry(
|
||||
note: dict[str, Any],
|
||||
*,
|
||||
include_content: bool = False,
|
||||
) -> dict[str, Any]:
|
||||
entry = {
|
||||
"note_id": note.get("note_id"),
|
||||
"title": note.get("title", ""),
|
||||
"category": note.get("category", "general"),
|
||||
"tags": note.get("tags", []),
|
||||
"created_at": note.get("created_at", ""),
|
||||
"updated_at": note.get("updated_at", ""),
|
||||
}
|
||||
wiki_filename = note.get("wiki_filename")
|
||||
if isinstance(wiki_filename, str) and wiki_filename:
|
||||
entry["wiki_filename"] = wiki_filename
|
||||
content = str(note.get("content", ""))
|
||||
if include_content:
|
||||
entry["content"] = content
|
||||
elif content:
|
||||
if len(content) > _DEFAULT_CONTENT_PREVIEW_CHARS:
|
||||
entry["content_preview"] = f"{content[:_DEFAULT_CONTENT_PREVIEW_CHARS].rstrip()}..."
|
||||
else:
|
||||
entry["content_preview"] = content
|
||||
return entry
|
||||
|
||||
|
||||
def _create_note_impl( # noqa: PLR0911
|
||||
title: str,
|
||||
content: str,
|
||||
category: str = "general",
|
||||
tags: list[str] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Create one note. Public — used by ``append_note_content`` and tests."""
|
||||
with _notes_lock:
|
||||
try:
|
||||
_ensure_notes_loaded()
|
||||
if not title or not title.strip():
|
||||
return {"success": False, "error": "Title cannot be empty", "note_id": None}
|
||||
if not content or not content.strip():
|
||||
return {"success": False, "error": "Content cannot be empty", "note_id": None}
|
||||
if category not in _VALID_NOTE_CATEGORIES:
|
||||
return {
|
||||
"success": False,
|
||||
"error": (
|
||||
f"Invalid category. Must be one of: {', '.join(_VALID_NOTE_CATEGORIES)}"
|
||||
),
|
||||
"note_id": None,
|
||||
}
|
||||
|
||||
note_id = ""
|
||||
for _ in range(20):
|
||||
candidate = str(uuid.uuid4())[:5]
|
||||
if candidate not in _notes_storage:
|
||||
note_id = candidate
|
||||
break
|
||||
if not note_id:
|
||||
return {"success": False, "error": "Failed to allocate note ID", "note_id": None}
|
||||
|
||||
timestamp = datetime.now(UTC).isoformat()
|
||||
note = {
|
||||
"title": title.strip(),
|
||||
"content": content.strip(),
|
||||
"category": category,
|
||||
"tags": tags or [],
|
||||
"created_at": timestamp,
|
||||
"updated_at": timestamp,
|
||||
}
|
||||
_notes_storage[note_id] = note
|
||||
_append_note_event("create", note_id, note)
|
||||
if category == "wiki":
|
||||
_persist_wiki_note(note_id, note)
|
||||
except (ValueError, TypeError) as e:
|
||||
return {"success": False, "error": f"Failed to create note: {e}", "note_id": None}
|
||||
except OSError as e:
|
||||
return {"success": False, "error": f"Failed to persist wiki note: {e}", "note_id": None}
|
||||
else:
|
||||
return {
|
||||
"success": True,
|
||||
"note_id": note_id,
|
||||
"message": f"Note '{title}' created successfully",
|
||||
}
|
||||
|
||||
|
||||
def _list_notes_impl(
|
||||
category: str | None = None,
|
||||
tags: list[str] | None = None,
|
||||
search: str | None = None,
|
||||
include_content: bool = False,
|
||||
) -> dict[str, Any]:
|
||||
with _notes_lock:
|
||||
try:
|
||||
filtered = _filter_notes(category=category, tags=tags, search_query=search)
|
||||
notes = [_to_note_listing_entry(n, include_content=include_content) for n in filtered]
|
||||
except (ValueError, TypeError) as e:
|
||||
return {
|
||||
"success": False,
|
||||
"error": f"Failed to list notes: {e}",
|
||||
"notes": [],
|
||||
"total_count": 0,
|
||||
}
|
||||
return {"success": True, "notes": notes, "total_count": len(notes)}
|
||||
|
||||
|
||||
def _get_note_impl(note_id: str) -> dict[str, Any]:
|
||||
with _notes_lock:
|
||||
try:
|
||||
_ensure_notes_loaded()
|
||||
if not note_id or not note_id.strip():
|
||||
return {"success": False, "error": "Note ID cannot be empty", "note": None}
|
||||
note = _notes_storage.get(note_id)
|
||||
if note is None:
|
||||
return {
|
||||
"success": False,
|
||||
"error": f"Note with ID '{note_id}' not found",
|
||||
"note": None,
|
||||
}
|
||||
note_with_id = note.copy()
|
||||
note_with_id["note_id"] = note_id
|
||||
except (ValueError, TypeError) as e:
|
||||
return {"success": False, "error": f"Failed to get note: {e}", "note": None}
|
||||
else:
|
||||
return {"success": True, "note": note_with_id}
|
||||
|
||||
|
||||
def _update_note_impl(
|
||||
note_id: str,
|
||||
title: str | None = None,
|
||||
content: str | None = None,
|
||||
tags: list[str] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
with _notes_lock:
|
||||
try:
|
||||
_ensure_notes_loaded()
|
||||
if note_id not in _notes_storage:
|
||||
return {"success": False, "error": f"Note with ID '{note_id}' not found"}
|
||||
note = _notes_storage[note_id]
|
||||
if title is not None:
|
||||
if not title.strip():
|
||||
return {"success": False, "error": "Title cannot be empty"}
|
||||
note["title"] = title.strip()
|
||||
if content is not None:
|
||||
if not content.strip():
|
||||
return {"success": False, "error": "Content cannot be empty"}
|
||||
note["content"] = content.strip()
|
||||
if tags is not None:
|
||||
note["tags"] = tags
|
||||
note["updated_at"] = datetime.now(UTC).isoformat()
|
||||
_append_note_event("update", note_id, note)
|
||||
if note.get("category") == "wiki":
|
||||
_persist_wiki_note(note_id, note)
|
||||
except (ValueError, TypeError) as e:
|
||||
return {"success": False, "error": f"Failed to update note: {e}"}
|
||||
except OSError as e:
|
||||
return {"success": False, "error": f"Failed to persist wiki note: {e}"}
|
||||
else:
|
||||
return {
|
||||
"success": True,
|
||||
"message": f"Note '{note['title']}' updated successfully",
|
||||
}
|
||||
|
||||
|
||||
def _delete_note_impl(note_id: str) -> dict[str, Any]:
|
||||
with _notes_lock:
|
||||
try:
|
||||
_ensure_notes_loaded()
|
||||
if note_id not in _notes_storage:
|
||||
return {"success": False, "error": f"Note with ID '{note_id}' not found"}
|
||||
note = _notes_storage[note_id]
|
||||
note_title = note["title"]
|
||||
if note.get("category") == "wiki":
|
||||
_remove_wiki_note(note_id, note)
|
||||
del _notes_storage[note_id]
|
||||
_append_note_event("delete", note_id)
|
||||
except (ValueError, TypeError) as e:
|
||||
return {"success": False, "error": f"Failed to delete note: {e}"}
|
||||
except OSError as e:
|
||||
return {"success": False, "error": f"Failed to delete wiki note: {e}"}
|
||||
else:
|
||||
return {
|
||||
"success": True,
|
||||
"message": f"Note '{note_title}' deleted successfully",
|
||||
}
|
||||
|
||||
|
||||
def append_note_content(note_id: str, delta: str) -> dict[str, Any]:
|
||||
"""Append text to an existing note's content. Used by the agents-graph
|
||||
wiki-update hook on agent_finish."""
|
||||
with _notes_lock:
|
||||
try:
|
||||
_ensure_notes_loaded()
|
||||
if note_id not in _notes_storage:
|
||||
return {"success": False, "error": f"Note with ID '{note_id}' not found"}
|
||||
note = _notes_storage[note_id]
|
||||
existing = str(note.get("content") or "")
|
||||
updated = f"{existing.rstrip()}{delta}"
|
||||
return _update_note_impl(note_id=note_id, content=updated)
|
||||
except (ValueError, TypeError) as e:
|
||||
return {"success": False, "error": f"Failed to append note content: {e}"}
|
||||
|
||||
|
||||
# --- public tools ---------------------------------------------------------
|
||||
|
||||
|
||||
@strix_tool(timeout=30)
|
||||
async def create_note(
|
||||
ctx: RunContextWrapper,
|
||||
@@ -35,26 +428,13 @@ async def create_note(
|
||||
"""Create a note in the current run's notes store.
|
||||
|
||||
Notes are persisted to ``run_dir/notes/notes.jsonl`` and (for the
|
||||
``wiki`` category) rendered as Markdown to ``run_dir/wiki/<slug>.md``.
|
||||
|
||||
Args:
|
||||
title: Required, non-empty title.
|
||||
content: Note body. Markdown is preserved.
|
||||
category: One of ``"general" | "findings" | "methodology" |
|
||||
"questions" | "plan" | "wiki"``.
|
||||
tags: Optional list of free-form tags.
|
||||
``wiki`` category) rendered as Markdown to
|
||||
``run_dir/wiki/<slug>.md``.
|
||||
"""
|
||||
# The legacy function does file I/O under a threading.RLock.
|
||||
# Wrap in to_thread so we don't block the event loop while waiting
|
||||
# on the lock or fsync.
|
||||
result = await asyncio.to_thread(
|
||||
_impl.create_note,
|
||||
title=title,
|
||||
content=content,
|
||||
category=category,
|
||||
tags=tags,
|
||||
del ctx
|
||||
return _dump(
|
||||
await asyncio.to_thread(_create_note_impl, title, content, category, tags),
|
||||
)
|
||||
return _dump(result)
|
||||
|
||||
|
||||
@strix_tool(timeout=30)
|
||||
@@ -65,30 +445,24 @@ async def list_notes(
|
||||
search: str | None = None,
|
||||
include_content: bool = False,
|
||||
) -> str:
|
||||
"""List notes, optionally filtered.
|
||||
|
||||
Args:
|
||||
category: Filter by category.
|
||||
tags: Filter to notes that have any of these tags.
|
||||
search: Substring match against title and content.
|
||||
include_content: When False (default), entries get a ``content_preview``;
|
||||
when True, full content is included.
|
||||
"""
|
||||
result = await asyncio.to_thread(
|
||||
_impl.list_notes,
|
||||
category=category,
|
||||
tags=tags,
|
||||
search=search,
|
||||
include_content=include_content,
|
||||
"""List notes, optionally filtered by category / tags / substring."""
|
||||
del ctx
|
||||
return _dump(
|
||||
await asyncio.to_thread(
|
||||
_list_notes_impl,
|
||||
category=category,
|
||||
tags=tags,
|
||||
search=search,
|
||||
include_content=include_content,
|
||||
),
|
||||
)
|
||||
return _dump(result)
|
||||
|
||||
|
||||
@strix_tool(timeout=30)
|
||||
async def get_note(ctx: RunContextWrapper, note_id: str) -> str:
|
||||
"""Fetch one note by its 5-char ID. Returns full content."""
|
||||
result = await asyncio.to_thread(_impl.get_note, note_id=note_id)
|
||||
return _dump(result)
|
||||
del ctx
|
||||
return _dump(await asyncio.to_thread(_get_note_impl, note_id))
|
||||
|
||||
|
||||
@strix_tool(timeout=30)
|
||||
@@ -99,19 +473,21 @@ async def update_note(
|
||||
content: str | None = None,
|
||||
tags: list[str] | None = None,
|
||||
) -> str:
|
||||
"""Update a note's title, content, or tags. Pass ``None`` to leave a field unchanged."""
|
||||
result = await asyncio.to_thread(
|
||||
_impl.update_note,
|
||||
note_id=note_id,
|
||||
title=title,
|
||||
content=content,
|
||||
tags=tags,
|
||||
"""Update a note's title, content, or tags."""
|
||||
del ctx
|
||||
return _dump(
|
||||
await asyncio.to_thread(
|
||||
_update_note_impl,
|
||||
note_id=note_id,
|
||||
title=title,
|
||||
content=content,
|
||||
tags=tags,
|
||||
),
|
||||
)
|
||||
return _dump(result)
|
||||
|
||||
|
||||
@strix_tool(timeout=30)
|
||||
async def delete_note(ctx: RunContextWrapper, note_id: str) -> str:
|
||||
"""Delete a note. For wiki notes, also removes the rendered Markdown file."""
|
||||
result = await asyncio.to_thread(_impl.delete_note, note_id=note_id)
|
||||
return _dump(result)
|
||||
del ctx
|
||||
return _dump(await asyncio.to_thread(_delete_note_impl, note_id))
|
||||
|
||||
Reference in New Issue
Block a user