Compare commits

...

1 Commits

Author SHA1 Message Date
claude[bot] 1b1f1c7428 fix: search_notes output_format='text' now returns markdown instead of raw JSON
Add _format_search_markdown() helper that formats SearchResponse as compact
human-readable markdown, following the same pattern used by build_context and
recent_activity. When output_format="text" (the default), search_notes now
returns a formatted markdown string instead of a Pydantic model object (which
was being serialized as JSON by the MCP server).

Update existing tests to use output_format="json" when they need to inspect
result structure, and update mock-based unit tests to assert isinstance(result, str).

Fixes #641

Co-authored-by: Paul Hernandez <phernandez@users.noreply.github.com>
Signed-off-by: claude[bot] <41898282+claude[bot]@users.noreply.github.com>
2026-03-02 16:59:38 +00:00
3 changed files with 134 additions and 45 deletions
+42 -2
View File
@@ -23,6 +23,46 @@ from basic_memory.schemas.search import (
)
def _format_search_markdown(result: SearchResponse, project: str) -> str:
"""Format SearchResponse as compact human-readable markdown.
Produces a readable representation suitable for LLM consumption,
mirroring the pattern used by build_context and recent_activity.
"""
if not result.results:
return f"No results found in project '{project}'."
lines = [f"# Search Results ({len(result.results)} found)"]
lines.append("")
for item in result.results:
lines.append(f"## {item.title}")
if item.permalink:
lines.append(f"permalink: {item.permalink}")
if item.file_path:
lines.append(f"file: {item.file_path}")
if item.matched_chunk:
lines.append("")
lines.append(item.matched_chunk)
elif item.content:
lines.append("")
# Truncate long content for readability
content = item.content
if len(content) > 500:
content = content[:497] + "..."
lines.append(content)
lines.append("")
# Pagination footer
page_info = f"page {result.current_page}"
if result.has_more:
page_info += f" | use page={result.current_page + 1} for more"
lines.append(f"---")
lines.append(f"*{len(result.results)} results | {page_info} | project: {project}*")
return "\n".join(lines)
def _semantic_search_enabled_for_text_search() -> bool:
"""Resolve semantic-search enablement in both MCP and CLI invocation paths."""
try:
@@ -273,7 +313,7 @@ async def search_notes(
status: Optional[str] = None,
min_similarity: Optional[float] = None,
context: Context | None = None,
) -> SearchResponse | dict | str:
) -> dict | str:
"""Search across all content in the knowledge base with comprehensive syntax support.
This tool searches the knowledge base using full-text search, pattern matching,
@@ -565,7 +605,7 @@ async def search_notes(
if output_format == "json":
return result.model_dump(mode="json", exclude_none=True)
return result
return _format_search_markdown(result, active_project.name)
except Exception as e:
logger.error(
+30
View File
@@ -18,6 +18,7 @@ from basic_memory.mcp.tools import (
recent_activity,
write_note,
)
from basic_memory.mcp.tools.search import search_notes
@pytest.mark.asyncio
@@ -361,3 +362,32 @@ async def test_build_context_json_default_and_text_mode(client, test_graph, test
)
assert isinstance(text_result, str)
assert "# Context:" in text_result
@pytest.mark.asyncio
async def test_search_notes_text_and_json_modes(app, test_project):
"""search_notes with output_format='text' must return markdown, not a JSON object."""
await write_note(
project=test_project.name,
title="Mode Search Note",
directory="mode-tests",
content="# Mode Search Note\n\nsearchable content for output mode test",
)
text_result = await search_notes(
query="searchable",
project=test_project.name,
output_format="text",
)
assert isinstance(text_result, str), "output_format='text' must return str, not a Pydantic model or dict"
assert "# Search Results" in text_result
assert "Mode Search Note" in text_result
json_result = await search_notes(
query="searchable",
project=test_project.name,
output_format="json",
)
assert isinstance(json_result, dict), "output_format='json' must return dict"
assert "results" in json_result
assert any(r["title"] == "Mode Search Note" for r in json_result["results"])
+62 -43
View File
@@ -23,14 +23,17 @@ async def test_search_text(client, test_project):
assert result
# Search for it
response = await search_notes(project=test_project.name, query="searchable")
response = await search_notes(
project=test_project.name, query="searchable", output_format="json"
)
# Verify results - handle both success and error cases
if isinstance(response, SearchResponse):
# Success case - verify SearchResponse
assert len(response.results) > 0
if isinstance(response, dict):
# Success case - verify SearchResponse dict
assert len(response["results"]) > 0
assert any(
r.permalink == f"{test_project.name}/test/test-search-note" for r in response.results
r["permalink"] == f"{test_project.name}/test/test-search-note"
for r in response["results"]
)
else:
# If search failed and returned error message, test should fail with informative message
@@ -52,7 +55,7 @@ async def test_search_title(client, test_project):
# Search for it
response = await search_notes(
project=test_project.name, query="Search Note", search_type="title"
project=test_project.name, query="Search Note", search_type="title", output_format="json"
)
# Verify results - handle both success and error cases
@@ -60,10 +63,11 @@ async def test_search_title(client, test_project):
# If search failed and returned error message, test should fail with informative message
pytest.fail(f"Search failed with error: {response}")
else:
# Success case - verify SearchResponse
assert len(response.results) > 0
# Success case - verify SearchResponse dict
assert len(response["results"]) > 0
assert any(
r.permalink == f"{test_project.name}/test/test-search-note" for r in response.results
r["permalink"] == f"{test_project.name}/test/test-search-note"
for r in response["results"]
)
@@ -85,14 +89,16 @@ async def test_search_permalink(client, test_project):
project=test_project.name,
query=f"{test_project.name}/test/test-search-note",
search_type="permalink",
output_format="json",
)
# Verify results - handle both success and error cases
if isinstance(response, SearchResponse):
# Success case - verify SearchResponse
assert len(response.results) > 0
if isinstance(response, dict):
# Success case - verify SearchResponse dict
assert len(response["results"]) > 0
assert any(
r.permalink == f"{test_project.name}/test/test-search-note" for r in response.results
r["permalink"] == f"{test_project.name}/test/test-search-note"
for r in response["results"]
)
else:
# If search failed and returned error message, test should fail with informative message
@@ -117,14 +123,16 @@ async def test_search_permalink_match(client, test_project):
project=test_project.name,
query=f"{test_project.name}/test/test-search-*",
search_type="permalink",
output_format="json",
)
# Verify results - handle both success and error cases
if isinstance(response, SearchResponse):
# Success case - verify SearchResponse
assert len(response.results) > 0
if isinstance(response, dict):
# Success case - verify SearchResponse dict
assert len(response["results"]) > 0
assert any(
r.permalink == f"{test_project.name}/test/test-search-note" for r in response.results
r["permalink"] == f"{test_project.name}/test/test-search-note"
for r in response["results"]
)
else:
# If search failed and returned error message, test should fail with informative message
@@ -142,13 +150,16 @@ async def test_search_memory_url_with_project_prefix(client, test_project):
)
assert result
response = await search_notes(query=f"memory://{test_project.name}/test/memory-url-search-note")
response = await search_notes(
query=f"memory://{test_project.name}/test/memory-url-search-note",
output_format="json",
)
if isinstance(response, SearchResponse):
assert len(response.results) > 0
if isinstance(response, dict):
assert len(response["results"]) > 0
assert any(
r.permalink == f"{test_project.name}/test/memory-url-search-note"
for r in response.results
r["permalink"] == f"{test_project.name}/test/memory-url-search-note"
for r in response["results"]
)
else:
pytest.fail(f"Search failed with error: {response}")
@@ -169,15 +180,16 @@ async def test_search_pagination(client, test_project):
# Search for it
response = await search_notes(
project=test_project.name, query="searchable", page=1, page_size=1
project=test_project.name, query="searchable", page=1, page_size=1, output_format="json"
)
# Verify results - handle both success and error cases
if isinstance(response, SearchResponse):
# Success case - verify SearchResponse
assert len(response.results) == 1
if isinstance(response, dict):
# Success case - verify SearchResponse dict
assert len(response["results"]) == 1
assert any(
r.permalink == f"{test_project.name}/test/test-search-note" for r in response.results
r["permalink"] == f"{test_project.name}/test/test-search-note"
for r in response["results"]
)
else:
# If search failed and returned error message, test should fail with informative message
@@ -196,12 +208,14 @@ async def test_search_with_type_filter(client, test_project):
)
# Search with note type filter
response = await search_notes(project=test_project.name, query="type", note_types=["note"])
response = await search_notes(
project=test_project.name, query="type", note_types=["note"], output_format="json"
)
# Verify results - handle both success and error cases
if isinstance(response, SearchResponse):
if isinstance(response, dict):
# Success case - verify all results are entities
assert all(r.type == "entity" for r in response.results)
assert all(r["type"] == "entity" for r in response["results"])
else:
# If search failed and returned error message, test should fail with informative message
pytest.fail(f"Search failed with error: {response}")
@@ -219,12 +233,14 @@ async def test_search_with_entity_type_filter(client, test_project):
)
# Search with entity_types (SearchItemType) filter
response = await search_notes(project=test_project.name, query="type", entity_types=["entity"])
response = await search_notes(
project=test_project.name, query="type", entity_types=["entity"], output_format="json"
)
# Verify results - handle both success and error cases
if isinstance(response, SearchResponse):
if isinstance(response, dict):
# Success case - verify all results are entities
assert all(r.type == "entity" for r in response.results)
assert all(r["type"] == "entity" for r in response["results"])
else:
# If search failed and returned error message, test should fail with informative message
pytest.fail(f"Search failed with error: {response}")
@@ -244,13 +260,16 @@ async def test_search_with_date_filter(client, test_project):
# Search with date filter
one_hour_ago = datetime.now() - timedelta(hours=1)
response = await search_notes(
project=test_project.name, query="recent", after_date=one_hour_ago.isoformat()
project=test_project.name,
query="recent",
after_date=one_hour_ago.isoformat(),
output_format="json",
)
# Verify results - handle both success and error cases
if isinstance(response, SearchResponse):
if isinstance(response, dict):
# Success case - verify we get results within timeframe
assert len(response.results) > 0
assert len(response["results"]) > 0
else:
# If search failed and returned error message, test should fail with informative message
pytest.fail(f"Search failed with error: {response}")
@@ -468,7 +487,7 @@ async def test_search_notes_sets_retrieval_mode_for_semantic_types(monkeypatch,
search_type=search_type,
)
assert isinstance(result, SearchResponse)
assert isinstance(result, str)
assert captured_payload["text"] == "semantic lookup"
# "semantic" is an alias for "vector" retrieval mode
expected_mode = "vector" if search_type in ("vector", "semantic") else search_type
@@ -563,7 +582,7 @@ async def test_search_notes_filter_only_metadata(monkeypatch):
metadata_filters={"status": "in-progress"},
)
assert isinstance(result, SearchResponse)
assert isinstance(result, str)
assert captured_payload["metadata_filters"] == {"status": "in-progress"}
# No text/title/permalink should be set
assert captured_payload.get("text") is None
@@ -605,7 +624,7 @@ async def test_search_notes_filter_only_tags(monkeypatch):
tags=["security", "oauth"],
)
assert isinstance(result, SearchResponse)
assert isinstance(result, str)
assert captured_payload["tags"] == ["security", "oauth"]
assert captured_payload.get("text") is None
@@ -1158,7 +1177,7 @@ async def test_search_notes_tag_prefix_converts_to_tags_filter(monkeypatch):
query="tag:security",
)
assert isinstance(result, SearchResponse)
assert isinstance(result, str)
assert captured_payload["tags"] == ["security"]
# No text query should be set — tag: prefix was consumed
assert captured_payload.get("text") is None
@@ -1199,7 +1218,7 @@ async def test_search_notes_tag_prefix_merges_with_explicit_tags(monkeypatch):
tags=["oauth"],
)
assert isinstance(result, SearchResponse)
assert isinstance(result, str)
assert set(captured_payload["tags"]) == {"security", "oauth"}
assert captured_payload.get("text") is None
@@ -1238,7 +1257,7 @@ async def test_search_notes_multiple_tag_prefixes(monkeypatch):
query="tag:coffee AND tag:brewing",
)
assert isinstance(result, SearchResponse)
assert isinstance(result, str)
assert set(captured_payload["tags"]) == {"coffee", "brewing"}
# Boolean connector AND should be stripped, leaving no text query
assert captured_payload.get("text") is None
@@ -1283,7 +1302,7 @@ async def test_search_notes_tag_prefix_with_remaining_text(monkeypatch):
query="authentication tag:security",
)
assert isinstance(result, SearchResponse)
assert isinstance(result, str)
assert captured_payload["tags"] == ["security"]
# Remaining text should be preserved as the query
assert captured_payload["text"] == "authentication"