admin/basicmachines-co-basic-memory
1
0
Fork 0
mirror of https://github.com/basicmachines-co/basic-memory synced 2026-06-21 13:47:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fix: Beta testing fixes (#16)

Browse Source 
Co-authored-by: phernandez <phernandez@basicmachines.co>
This commit is contained in:
Paul Hernandez
2025-02-28 20:23:53 -06:00
committed by GitHub
parent 4bcbeacdd2
commit f8078cdd46
49 changed files with 1075 additions and 1215 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/basic_memory/alembic/versions/cc7172b46608_update_search_index_schema.py
+106
View File
@@ -0,0 +1,106 @@
"""Update search index schema
Revision ID: cc7172b46608
Revises: 502b60eaa905
Create Date: 2025-02-28 18:48:23.244941
"""
from typing import Sequence, Union
from alembic import op
# revision identifiers, used by Alembic.
revision: str = "cc7172b46608"
down_revision: Union[str, None] = "502b60eaa905"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
"""Upgrade database schema to use new search index with content_stems and content_snippet."""
# First, drop the existing search_index table
op.execute("DROP TABLE IF EXISTS search_index")
# Create new search_index with updated schema
op.execute("""
CREATE VIRTUAL TABLE IF NOT EXISTS search_index USING fts5(
-- Core entity fields
id UNINDEXED, -- Row ID
title, -- Title for searching
content_stems, -- Main searchable content split into stems
content_snippet, -- File content snippet for display
permalink, -- Stable identifier (now indexed for path search)
file_path UNINDEXED, -- Physical location
type UNINDEXED, -- entity/relation/observation
-- Relation fields
from_id UNINDEXED, -- Source entity
to_id UNINDEXED, -- Target entity
relation_type UNINDEXED, -- Type of relation
-- Observation fields
entity_id UNINDEXED, -- Parent entity
category UNINDEXED, -- Observation category
-- Common fields
metadata UNINDEXED, -- JSON metadata
created_at UNINDEXED, -- Creation timestamp
updated_at UNINDEXED, -- Last update
-- Configuration
tokenize='unicode61 tokenchars 0x2F', -- Hex code for /
prefix='1,2,3,4' -- Support longer prefixes for paths
);
""")
# Print instruction to manually reindex after migration
print("\n------------------------------------------------------------------")
print("IMPORTANT: After migration completes, manually run the reindex command:")
print("basic-memory sync")
print("------------------------------------------------------------------\n")
def downgrade() -> None:
"""Downgrade database schema to use old search index."""
# Drop the updated search_index table
op.execute("DROP TABLE IF EXISTS search_index")
# Recreate the original search_index schema
op.execute("""
CREATE VIRTUAL TABLE IF NOT EXISTS search_index USING fts5(
-- Core entity fields
id UNINDEXED, -- Row ID
title, -- Title for searching
content, -- Main searchable content
permalink, -- Stable identifier (now indexed for path search)
file_path UNINDEXED, -- Physical location
type UNINDEXED, -- entity/relation/observation
-- Relation fields
from_id UNINDEXED, -- Source entity
to_id UNINDEXED, -- Target entity
relation_type UNINDEXED, -- Type of relation
-- Observation fields
entity_id UNINDEXED, -- Parent entity
category UNINDEXED, -- Observation category
-- Common fields
metadata UNINDEXED, -- JSON metadata
created_at UNINDEXED, -- Creation timestamp
updated_at UNINDEXED, -- Last update
-- Configuration
tokenize='unicode61 tokenchars 0x2F', -- Hex code for /
prefix='1,2,3,4' -- Support longer prefixes for paths
);
""")
# Print instruction to manually reindex after migration
print("\n------------------------------------------------------------------")
print("IMPORTANT: After downgrade completes, manually run the reindex command:")
print("basic-memory sync")
print("------------------------------------------------------------------\n")
src/basic_memory/api/routers/memory_router.py
+3 -2
View File
@@ -32,6 +32,7 @@ async def to_graph_context(context, entity_repository: EntityRepository, page: i
return EntitySummary(
title=item.title, # pyright: ignore
permalink=item.permalink,
content=item.content,
file_path=item.file_path,
created_at=item.created_at,
)
@@ -52,8 +53,8 @@ async def to_graph_context(context, entity_repository: EntityRepository, page: i
file_path=item.file_path,
permalink=item.permalink, # pyright: ignore
relation_type=item.type,
from_id=from_entity.permalink, # pyright: ignore
to_id=to_entity.permalink if to_entity else None,
from_entity=from_entity.permalink, # pyright: ignore
to_entity=to_entity.permalink if to_entity else None,
created_at=item.created_at,
)
case _: # pragma: no cover
src/basic_memory/api/routers/search_router.py
+22 -4
View File
@@ -1,11 +1,9 @@
"""Router for search operations."""
from dataclasses import asdict
from fastapi import APIRouter, BackgroundTasks
from basic_memory.schemas.search import SearchQuery, SearchResult, SearchResponse
from basic_memory.deps import SearchServiceDep
from basic_memory.deps import SearchServiceDep, EntityServiceDep
router = APIRouter(prefix="/search", tags=["search"])
@@ -14,6 +12,7 @@ router = APIRouter(prefix="/search", tags=["search"])
async def search(
query: SearchQuery,
search_service: SearchServiceDep,
entity_service: EntityServiceDep,
page: int = 1,
page_size: int = 10,
):
@@ -21,7 +20,26 @@ async def search(
limit = page_size
offset = (page - 1) * page_size
results = await search_service.search(query, limit=limit, offset=offset)
search_results = [SearchResult.model_validate(asdict(r)) for r in results]
search_results = []
for r in results:
entities = await entity_service.get_entities_by_id([r.entity_id, r.from_id, r.to_id]) # pyright: ignore
search_results.append(
SearchResult(
title=r.title, # pyright: ignore
type=r.type, # pyright: ignore
permalink=r.permalink,
score=r.score, # pyright: ignore
entity=entities[0].permalink if entities else None,
content=r.content,
file_path=r.file_path,
metadata=r.metadata,
category=r.category,
from_entity=entities[0].permalink if entities else None,
to_entity=entities[1].permalink if len(entities) > 1 else None,
relation_type=r.relation_type,
)
)
return SearchResponse(
results=search_results,
current_page=page,
src/basic_memory/cli/commands/tools.py
+15 -16
View File
@@ -9,7 +9,6 @@ from rich import print as rprint
from basic_memory.cli.app import app
from basic_memory.mcp.tools import build_context as mcp_build_context
from basic_memory.mcp.tools import get_entity as mcp_get_entity
from basic_memory.mcp.tools import read_note as mcp_read_note
from basic_memory.mcp.tools import recent_activity as mcp_recent_activity
from basic_memory.mcp.tools import search as mcp_search
@@ -79,7 +78,11 @@ def build_context(
max_related=max_related,
)
)
rprint(context.model_dump_json(indent=2))
# Use json module for more controlled serialization
import json
context_dict = context.model_dump(exclude_none=True)
print(json.dumps(context_dict, indent=2, ensure_ascii=True, default=str))
except Exception as e: # pragma: no cover
if not isinstance(e, typer.Exit):
typer.echo(f"Error during build_context: {e}", err=True)
@@ -107,7 +110,11 @@ def recent_activity(
max_related=max_related,
)
)
rprint(context.model_dump_json(indent=2))
# Use json module for more controlled serialization
import json
context_dict = context.model_dump(exclude_none=True)
print(json.dumps(context_dict, indent=2, ensure_ascii=True, default=str))
except Exception as e: # pragma: no cover
if not isinstance(e, typer.Exit):
typer.echo(f"Error during build_context: {e}", err=True)
@@ -139,7 +146,11 @@ def search(
after_date=after_date,
)
results = asyncio.run(mcp_search(query=search_query, page=page, page_size=page_size))
rprint(results.model_dump_json(indent=2))
# Use json module for more controlled serialization
import json
results_dict = results.model_dump(exclude_none=True)
print(json.dumps(results_dict, indent=2, ensure_ascii=True, default=str))
except Exception as e: # pragma: no cover
if not isinstance(e, typer.Exit):
logger.exception("Error during search", e)
@@ -148,18 +159,6 @@ def search(
raise
@tool_app.command()
def get_entity(identifier: str):
try:
entity = asyncio.run(mcp_get_entity(identifier=identifier))
rprint(entity.model_dump_json(indent=2))
except Exception as e: # pragma: no cover
if not isinstance(e, typer.Exit):
typer.echo(f"Error during get_entity: {e}", err=True)
raise typer.Exit(1)
raise
@tool_app.command(name="continue-conversation")
def continue_conversation(
topic: Annotated[Optional[str], typer.Option(help="Topic or keyword to search for")] = None,
src/basic_memory/mcp/main.py
+1 -2
View File
@@ -17,9 +17,8 @@ import basic_memory.mcp.tools # noqa: F401 # pragma: no cover
import basic_memory.mcp.prompts # noqa: F401 # pragma: no cover
if __name__ == "__main__": # pragma: no cover
home_dir = config.home
logger.info("Starting Basic Memory MCP server")
logger.info(f"Home directory: {home_dir}")
mcp.run()
mcp.run()
src/basic_memory/mcp/prompts/__init__.py
+7 -1
View File
@@ -12,4 +12,10 @@ from basic_memory.mcp.prompts import search
from basic_memory.mcp.prompts import ai_assistant_guide
from basic_memory.mcp.prompts import json_canvas_spec
__all__ = ["ai_assistant_guide", "continue_conversation", "json_canvas_spec", "recent_activity", "search"]
__all__ = [
"ai_assistant_guide",
"continue_conversation",
"json_canvas_spec",
"recent_activity",
"search",
]
src/basic_memory/mcp/prompts/ai_assistant_guide.py
+1 -1
View File
@@ -8,7 +8,7 @@ from basic_memory.mcp.server import mcp
@mcp.resource(
uri="memory://ai_assistant_guide",
name="ai_assistant_guide",
name="ai assistant guide",
description="Give an AI assistant guidance on how to use Basic Memory tools effectively",
)
def ai_assistant_guide() -> str:
src/basic_memory/mcp/prompts/continue_conversation.py
+19 -9
View File
@@ -12,15 +12,16 @@ import logfire
from pydantic import Field
from basic_memory.mcp.server import mcp
from basic_memory.mcp.tools.memory import build_context, recent_activity
from basic_memory.mcp.tools.build_context import build_context
from basic_memory.mcp.tools.recent_activity import recent_activity
from basic_memory.mcp.tools.search import search
from basic_memory.schemas.base import TimeFrame
from basic_memory.schemas.memory import GraphContext
from basic_memory.schemas.search import SearchQuery
from basic_memory.schemas.search import SearchQuery, SearchItemType
@mcp.prompt(
name="continue_conversation",
name="continue conversation",
description="Continue a previous conversation",
)
async def continue_conversation(
@@ -47,16 +48,19 @@ async def continue_conversation(
# If topic provided, search for it
if topic:
search_results = await search(SearchQuery(text=topic, after_date=timeframe))
search_results = await search(
SearchQuery(text=topic, after_date=timeframe, types=[SearchItemType.ENTITY])
)
# Build context from top results
# Build context from results
contexts = []
for result in search_results.results[:3]:
for result in search_results.results:
if hasattr(result, "permalink") and result.permalink:
context = await build_context(f"memory://{result.permalink}")
contexts.append(context)
return format_continuation_context(topic, contexts, timeframe)
# get context for the top 3 results
return format_continuation_context(topic, contexts[:3], timeframe)
# If no topic, get recent activity
recent = await recent_activity(timeframe=timeframe)
@@ -123,6 +127,12 @@ def format_continuation_context(
if hasattr(primary, "created_at"):
section += f"- **Created**: {primary.created_at.strftime('%Y-%m-%d %H:%M')}\n"
# Add content snippet
if hasattr(primary, "content") and primary.content: # pyright: ignore
content = primary.content or "" # pyright: ignore
if content:
section += f"- **Content Snippet**: {content}\n"
section += dedent(f"""
You can read this document with: `read_note("{primary.permalink}")`
@@ -145,8 +155,8 @@ def format_continuation_context(
display_type = rel_type.replace("_", " ").title()
section += f"- **{display_type}**:\n"
for rel in relations[:3]: # Limit to avoid overwhelming
if hasattr(rel, "to_id") and rel.to_id:
section += f" - `{rel.to_id}`\n"
if hasattr(rel, "to_entity") and rel.to_entity:
section += f" - `{rel.to_entity}`\n"
sections.append(section)
src/basic_memory/mcp/prompts/json_canvas_spec.py
+7 -5
View File
@@ -8,18 +8,20 @@ from basic_memory.mcp.server import mcp
@mcp.resource(
uri="memory://json_canvas_spec",
name="json_canvas_spec",
description="JSON Canvas specification for visualizing knowledge graphs in Obsidian"
name="json canvas spec",
description="JSON Canvas specification for visualizing knowledge graphs in Obsidian",
)
def json_canvas_spec() -> str:
"""Return the JSON Canvas specification for Obsidian visualizations.
Returns:
The JSON Canvas specification document.
"""
with logfire.span("Getting JSON Canvas spec"): # pyright: ignore
logger.info("Loading JSON Canvas spec resource")
canvas_spec = Path(__file__).parent.parent.parent.parent.parent / "data/json_canvas_spec_1_0.md"
canvas_spec = (
Path(__file__).parent.parent.parent.parent.parent / "data/json_canvas_spec_1_0.md"
)
content = canvas_spec.read_text()
logger.info(f"Loaded JSON Canvas spec ({len(content)} chars)")
return content
return content
src/basic_memory/mcp/prompts/recent_activity.py
+2 -2
View File
@@ -11,12 +11,12 @@ from pydantic import Field
from basic_memory.mcp.prompts.utils import format_context_summary
from basic_memory.mcp.server import mcp
from basic_memory.mcp.tools.memory import recent_activity as recent_activity_tool
from basic_memory.mcp.tools.recent_activity import recent_activity as recent_activity_tool
from basic_memory.schemas.base import TimeFrame
@mcp.prompt(
name="recent_activity",
name="recent activity",
description="Get recent activity from across the knowledge base",
)
async def recent_activity_prompt(
src/basic_memory/mcp/server.py
+1 -1
View File
@@ -8,4 +8,4 @@ configure_logging(level="INFO")
# Create the shared server instance
mcp = FastMCP("Basic Memory")
mcp = FastMCP("Basic Memory")
src/basic_memory/mcp/tools/__init__.py
+12 -23
View File
@@ -6,33 +6,22 @@ all tools with the MCP server.
"""
# Import tools to register them with MCP
from basic_memory.mcp.tools.resource import read_resource
from basic_memory.mcp.tools.memory import build_context, recent_activity
from basic_memory.mcp.tools.notes import read_note, write_note
from basic_memory.mcp.tools.delete_note import delete_note
from basic_memory.mcp.tools.read_file import read_file
from basic_memory.mcp.tools.build_context import build_context
from basic_memory.mcp.tools.recent_activity import recent_activity
from basic_memory.mcp.tools.read_note import read_note
from basic_memory.mcp.tools.write_note import write_note
from basic_memory.mcp.tools.search import search
from basic_memory.mcp.tools.canvas import canvas
from basic_memory.mcp.tools.knowledge import (
delete_entities,
get_entity,
get_entities,
)
__all__ = [
# Knowledge graph tools
"delete_entities",
"get_entity",
"get_entities",
# Search tools
"search",
# memory tools
"build_context",
"recent_activity",
# notes
"read_note",
"write_note",
# files
"read_resource",
# canvas
"canvas",
"delete_note",
"read_file",
"read_note",
"recent_activity",
"search",
"write_note",
]
src/basic_memory/mcp/tools/build_context.py
+87
View File
@@ -0,0 +1,87 @@
"""Build context tool for Basic Memory MCP server."""
from typing import Optional
import logfire
from loguru import logger
from basic_memory.mcp.async_client import client
from basic_memory.mcp.server import mcp
from basic_memory.mcp.tools.utils import call_get
from basic_memory.schemas.memory import (
GraphContext,
MemoryUrl,
memory_url_path,
normalize_memory_url,
)
from basic_memory.schemas.base import TimeFrame
@mcp.tool(
description="""Build context from a memory:// URI to continue conversations naturally.
Use this to follow up on previous discussions or explore related topics.
Timeframes support natural language like:
- "2 days ago"
- "last week"
- "today"
- "3 months ago"
Or standard formats like "7d", "24h"
""",
)
async def build_context(
url: MemoryUrl,
depth: Optional[int] = 1,
timeframe: Optional[TimeFrame] = "7d",
page: int = 1,
page_size: int = 10,
max_related: int = 10,
) -> GraphContext:
"""Get context needed to continue a discussion.
This tool enables natural continuation of discussions by loading relevant context
from memory:// URIs. It uses pattern matching to find relevant content and builds
a rich context graph of related information.
Args:
url: memory:// URI pointing to discussion content (e.g. memory://specs/search)
depth: How many relation hops to traverse (1-3 recommended for performance)
timeframe: How far back to look. Supports natural language like "2 days ago", "last week"
page: Page number of results to return (default: 1)
page_size: Number of results to return per page (default: 10)
max_related: Maximum number of related results to return (default: 10)
Returns:
GraphContext containing:
- primary_results: Content matching the memory:// URI
- related_results: Connected content via relations
- metadata: Context building details
Examples:
# Continue a specific discussion
build_context("memory://specs/search")
# Get deeper context about a component
build_context("memory://components/memory-service", depth=2)
# Look at recent changes to a specification
build_context("memory://specs/document-format", timeframe="today")
# Research the history of a feature
build_context("memory://features/knowledge-graph", timeframe="3 months ago")
"""
with logfire.span("Building context", url=url, depth=depth, timeframe=timeframe): # pyright: ignore [reportGeneralTypeIssues]
logger.info(f"Building context from {url}")
url = normalize_memory_url(url)
response = await call_get(
client,
f"/memory/{memory_url_path(url)}",
params={
"depth": depth,
"timeframe": timeframe,
"page": page,
"page_size": page_size,
"max_related": max_related,
},
)
return GraphContext.model_validate(response.json())
src/basic_memory/mcp/tools/delete_note.py
+31
View File
@@ -0,0 +1,31 @@
import logfire
from basic_memory.mcp.tools.utils import call_delete
from basic_memory.mcp.server import mcp
from basic_memory.mcp.async_client import client
from basic_memory.schemas import DeleteEntitiesResponse
@mcp.tool(description="Delete a note by title or permalink")
async def delete_note(identifier: str) -> bool:
"""Delete a note from the knowledge base.
Args:
identifier: Note title or permalink
Returns:
True if note was deleted, False otherwise
Examples:
# Delete by title
delete_note("Meeting Notes: Project Planning")
# Delete by permalink
delete_note("notes/project-planning")
"""
with logfire.span("Deleting note", identifier=identifier): # pyright: ignore [reportGeneralTypeIssues]
response = await call_delete(client, f"/knowledge/entities/{identifier}")
result = DeleteEntitiesResponse.model_validate(response.json())
return result.deleted
src/basic_memory/mcp/tools/knowledge.py
-68
View File
@@ -1,68 +0,0 @@
"""Knowledge graph management tools for Basic Memory MCP server."""
import logfire
from basic_memory.mcp.server import mcp
from basic_memory.mcp.tools.utils import call_get, call_post
from basic_memory.schemas.memory import memory_url_path
from basic_memory.schemas.request import (
GetEntitiesRequest,
)
from basic_memory.schemas.delete import (
DeleteEntitiesRequest,
)
from basic_memory.schemas.response import EntityListResponse, EntityResponse, DeleteEntitiesResponse
from basic_memory.mcp.async_client import client
@mcp.tool(
description="Get complete information about a specific entity including observations and relations",
)
async def get_entity(identifier: str) -> EntityResponse:
"""Get a specific entity info by its permalink.
Args:
identifier: Path identifier for the entity
"""
with logfire.span("Getting entity", permalink=identifier): # pyright: ignore [reportGeneralTypeIssues]
permalink = memory_url_path(identifier)
url = f"/knowledge/entities/{permalink}"
response = await call_get(client, url)
return EntityResponse.model_validate(response.json())
@mcp.tool(
description="Load multiple entities by their permalinks in a single request",
)
async def get_entities(request: GetEntitiesRequest) -> EntityListResponse:
"""Load multiple entities by their permalinks.
Args:
request: OpenNodesRequest containing list of permalinks to load
Returns:
EntityListResponse containing complete details for each requested entity
"""
with logfire.span("Getting multiple entities", permalink_count=len(request.permalinks)): # pyright: ignore [reportGeneralTypeIssues]
url = "/knowledge/entities"
response = await call_get(
client,
url,
params=[
("permalink", memory_url_path(identifier)) for identifier in request.permalinks
],
)
return EntityListResponse.model_validate(response.json())
@mcp.tool(
description="Permanently delete entities and all related content (observations and relations)",
)
async def delete_entities(request: DeleteEntitiesRequest) -> DeleteEntitiesResponse:
"""Delete entities from the knowledge graph."""
with logfire.span("Deleting entities", permalink_count=len(request.permalinks)): # pyright: ignore [reportGeneralTypeIssues]
url = "/knowledge/entities/delete"
request.permalinks = [memory_url_path(permlink) for permlink in request.permalinks]
response = await call_post(client, url, json=request.model_dump())
return DeleteEntitiesResponse.model_validate(response.json())
src/basic_memory/mcp/tools/resource.py → src/basic_memory/mcp/tools/read_file.py
+42 -5
View File
@@ -1,3 +1,10 @@
"""File reading tool for Basic Memory MCP server.
This module provides tools for reading raw file content directly,
supporting various file types including text, images, and other binary files.
Files are read directly without any knowledge graph processing.
"""
from loguru import logger
from basic_memory.mcp.server import mcp
@@ -136,10 +143,40 @@ def optimize_image(img, content_length, max_output_bytes=350000):
return buf.getvalue()
@mcp.tool(description="Read a single file's content by path or permalink")
async def read_resource(path: str) -> dict:
"""Get a file's raw content."""
logger.info("Reading resource", path=path)
@mcp.tool(description="Read a file's raw content by path or permalink")
async def read_file(path: str) -> dict:
"""Read a file's raw content by path or permalink.
This tool provides direct access to file content in the knowledge base,
handling different file types appropriately:
- Text files (markdown, code, etc.) are returned as plain text
- Images are automatically resized/optimized for display
- Other binary files are returned as base64 if below size limits
Args:
path: The path or permalink to the file. Can be:
- A regular file path (docs/example.md)
- A memory URL (memory://docs/example)
- A permalink (docs/example)
Returns:
A dictionary with the file content and metadata:
- For text: {"type": "text", "text": "content", "content_type": "text/markdown", "encoding": "utf-8"}
- For images: {"type": "image", "source": {"type": "base64", "media_type": "image/jpeg", "data": "base64_data"}}
- For other files: {"type": "document", "source": {"type": "base64", "media_type": "content_type", "data": "base64_data"}}
- For errors: {"type": "error", "error": "error message"}
Examples:
# Read a markdown file
result = await read_file("docs/project-specs.md")
# Read an image
image_data = await read_file("assets/diagram.png")
# Read using memory URL
content = await read_file("memory://docs/architecture")
"""
logger.info("Reading file", path=path)
url = memory_url_path(path)
response = await call_get(client, f"/resource/{url}")
@@ -176,7 +213,7 @@ async def read_resource(path: str) -> dict:
# Handle other file types
else:
logger.debug(f"Processing binary resource content_type {content_type}")
if content_length > 350000:
if content_length > 350000: # pragma: no cover
logger.warning("Document too large for response", size=content_length)
return {
"type": "error",
src/basic_memory/mcp/tools/read_note.py
+60
View File
@@ -0,0 +1,60 @@
"""Read note tool for Basic Memory MCP server."""
import logfire
from loguru import logger
from basic_memory.mcp.async_client import client
from basic_memory.mcp.server import mcp
from basic_memory.mcp.tools.utils import call_get
from basic_memory.schemas.memory import memory_url_path
@mcp.tool(
description="Read a markdown note by title or permalink.",
)
async def read_note(identifier: str, page: int = 1, page_size: int = 10) -> str:
"""Read a markdown note from the knowledge base.
This tool finds and retrieves a note by its title or permalink, returning
the raw markdown content including observations, relations, and metadata.
Unlike read_file, this tool is aware of the knowledge graph structure and
will attempt to resolve entity references if the file path doesn't exist.
Args:
identifier: The title or permalink of the note to read
Can be a full memory:// URL, a permalink, or a title
page: Page number for paginated results (default: 1)
page_size: Number of items per page (default: 10)
Returns:
The full markdown content of the note, either from file content
or constructed from entity data if direct file access fails.
For entities without markdown content, returns a message indicating
the entity was found but has no content.
Examples:
# Read by permalink
read_note("specs/search-spec")
# Read by title
read_note("Search Specification")
# Read with memory URL
read_note("memory://specs/search-spec")
# Read with pagination
read_note("Project Updates", page=2, page_size=5)
"""
with logfire.span("Reading note", identifier=identifier): # pyright: ignore [reportGeneralTypeIssues]
# Get the file via REST API
entity_path = memory_url_path(identifier)
path = f"/resource/{entity_path}"
logger.info(f"Reading note from URL: {path}")
response = await call_get(client, path, params={"page": page, "page_size": page_size})
# Just return the content as a string
if response.status_code == 200:
return response.text
else:
return f"Error: Could not find entity at {identifier}"
src/basic_memory/mcp/tools/memory.py → src/basic_memory/mcp/tools/recent_activity.py
+3 -78
View File
@@ -1,93 +1,18 @@
"""Discussion context tools for Basic Memory MCP server."""
"""Recent activity tool for Basic Memory MCP server."""
from typing import Optional, List
from loguru import logger
import logfire
from loguru import logger
from basic_memory.mcp.async_client import client
from basic_memory.mcp.server import mcp
from basic_memory.mcp.tools.utils import call_get
from basic_memory.schemas.memory import (
GraphContext,
MemoryUrl,
memory_url_path,
normalize_memory_url,
)
from basic_memory.schemas.memory import GraphContext
from basic_memory.schemas.base import TimeFrame
from basic_memory.schemas.search import SearchItemType
@mcp.tool(
description="""Build context from a memory:// URI to continue conversations naturally.
Use this to follow up on previous discussions or explore related topics.
Timeframes support natural language like:
- "2 days ago"
- "last week"
- "today"
- "3 months ago"
Or standard formats like "7d", "24h"
""",
)
async def build_context(
url: MemoryUrl,
depth: Optional[int] = 1,
timeframe: Optional[TimeFrame] = "7d",
page: int = 1,
page_size: int = 10,
max_related: int = 10,
) -> GraphContext:
"""Get context needed to continue a discussion.
This tool enables natural continuation of discussions by loading relevant context
from memory:// URIs. It uses pattern matching to find relevant content and builds
a rich context graph of related information.
Args:
url: memory:// URI pointing to discussion content (e.g. memory://specs/search)
depth: How many relation hops to traverse (1-3 recommended for performance)
timeframe: How far back to look. Supports natural language like "2 days ago", "last week"
page: Page number of results to return (default: 1)
page_size: Number of results to return per page (default: 10)
max_related: Maximum number of related results to return (default: 10)
Returns:
GraphContext containing:
- primary_results: Content matching the memory:// URI
- related_results: Connected content via relations
- metadata: Context building details
Examples:
# Continue a specific discussion
build_context("memory://specs/search")
# Get deeper context about a component
build_context("memory://components/memory-service", depth=2)
# Look at recent changes to a specification
build_context("memory://specs/document-format", timeframe="today")
# Research the history of a feature
build_context("memory://features/knowledge-graph", timeframe="3 months ago")
"""
with logfire.span("Building context", url=url, depth=depth, timeframe=timeframe): # pyright: ignore [reportGeneralTypeIssues]
logger.info(f"Building context from {url}")
url = normalize_memory_url(url)
response = await call_get(
client,
f"/memory/{memory_url_path(url)}",
params={
"depth": depth,
"timeframe": timeframe,
"page": page,
"page_size": page_size,
"max_related": max_related,
},
)
return GraphContext.model_validate(response.json())
@mcp.tool(
description="""Get recent activity from across the knowledge base.
src/basic_memory/mcp/tools/search.py
+45 -7
View File
@@ -15,17 +15,55 @@ from basic_memory.mcp.async_client import client
async def search(query: SearchQuery, page: int = 1, page_size: int = 10) -> SearchResponse:
"""Search across all content in basic-memory.
This tool searches the knowledge base using full-text search, pattern matching,
or exact permalink lookup. It supports filtering by content type, entity type,
and date.
Args:
query: SearchQuery object with search parameters including:
- text: Search text (required)
- types: Optional list of content types to search ("document" or "entity")
- entity_types: Optional list of entity types to filter by
- after_date: Optional date filter for recent content
page: the page number of results to return (default 1)
page_size: the number of results to return per page (default 10)
- text: Full-text search (e.g., "project planning")
- title: Search only in titles (e.g., "Meeting notes")
- permalink: Exact permalink match (e.g., "docs/meeting-notes")
- permalink_match: Pattern matching for permalinks (e.g., "docs/*-notes")
- types: Optional list of content types to search (e.g., ["entity", "observation"])
- entity_types: Optional list of entity types to filter by (e.g., ["note", "person"])
- after_date: Optional date filter for recent content (e.g., "1 week", "2d")
page: The page number of results to return (default 1)
page_size: The number of results to return per page (default 10)
Returns:
SearchResponse with search results and metadata
SearchResponse with:
- results: List of matching SearchResult objects with:
- id: Internal ID
- title: Document/entity title
- type: Content type (entity, observation, relation)
- score: Relevance score (higher = more relevant)
- permalink: Permalink for accessing the content
- file_path: File path on disk
- metadata: Additional metadata about the result
- current_page: Current page number
- page_size: Number of results per page
Examples:
# Basic text search
results = await search(SearchQuery(text="project planning"))
# Search with type filter
results = await search(SearchQuery(
text="meeting notes",
types=["entity"],
))
# Search for recent content
results = await search(SearchQuery(
text="bug report",
after_date="1 week"
))
# Pattern matching on permalinks
results = await search(SearchQuery(
permalink_match="docs/meeting-*"
))
"""
with logfire.span("Searching for {query}", query=query): # pyright: ignore [reportGeneralTypeIssues]
logger.info(f"Searching for {query}")
src/basic_memory/mcp/tools/utils.py
+245 -17
View File
@@ -1,3 +1,9 @@
"""Utility functions for making HTTP requests in Basic Memory MCP tools.
These functions provide a consistent interface for making HTTP requests
to the Basic Memory API, with improved error handling and logging.
"""
import typing
from httpx import Response, URL, AsyncClient, HTTPStatusError
@@ -17,6 +23,54 @@ from loguru import logger
from mcp.server.fastmcp.exceptions import ToolError
def get_error_message(status_code: int, url: URL | str, method: str) -> str:
"""Get a friendly error message based on the HTTP status code.
Args:
status_code: The HTTP status code
url: The URL that was requested
method: The HTTP method used
Returns:
A user-friendly error message
"""
# Extract path from URL for cleaner error messages
if isinstance(url, str):
path = url.split("/")[-1]
else:
path = str(url).split("/")[-1] if url else "resource"
# Client errors (400-499)
if status_code == 400:
return f"Invalid request: The request to '{path}' was malformed or invalid"
elif status_code == 401: # pragma: no cover
return f"Authentication required: You need to authenticate to access '{path}'"
elif status_code == 403: # pragma: no cover
return f"Access denied: You don't have permission to access '{path}'"
elif status_code == 404:
return f"Resource not found: '{path}' doesn't exist or has been moved"
elif status_code == 409: # pragma: no cover
return f"Conflict: The request for '{path}' conflicts with the current state"
elif status_code == 429: # pragma: no cover
return "Too many requests: Please slow down and try again later"
elif 400 <= status_code < 500: # pragma: no cover
return f"Client error ({status_code}): The request for '{path}' could not be completed"
# Server errors (500-599)
elif status_code == 500:
return f"Internal server error: Something went wrong processing '{path}'"
elif status_code == 503: # pragma: no cover
return (
f"Service unavailable: The server is currently unable to handle requests for '{path}'"
)
elif 500 <= status_code < 600: # pragma: no cover
return f"Server error ({status_code}): The server encountered an error handling '{path}'"
# Fallback for any other status code
else: # pragma: no cover
return f"HTTP error {status_code}: {method} request to '{path}' failed"
async def call_get(
client: AsyncClient,
url: URL | str,
@@ -29,6 +83,25 @@ async def call_get(
timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
extensions: RequestExtensions | None = None,
) -> Response:
"""Make a GET request and handle errors appropriately.
Args:
client: The HTTPX AsyncClient to use
url: The URL to request
params: Query parameters
headers: HTTP headers
cookies: HTTP cookies
auth: Authentication
follow_redirects: Whether to follow redirects
timeout: Request timeout
extensions: HTTPX extensions
Returns:
The HTTP response
Raises:
ToolError: If the request fails with an appropriate error message
"""
logger.debug(f"Calling GET '{url}' params: '{params}'")
try:
response = await client.get(
@@ -41,11 +114,33 @@ async def call_get(
timeout=timeout,
extensions=extensions,
)
response.raise_for_status()
return response
if response.is_success:
return response
# Handle different status codes differently
status_code = response.status_code
error_message = get_error_message(status_code, url, "GET")
# Log at appropriate level based on status code
if 400 <= status_code < 500:
# Client errors: log as info except for 429 (Too Many Requests)
if status_code == 429: # pragma: no cover
logger.warning(f"Rate limit exceeded: GET {url}: {error_message}")
else:
logger.info(f"Client error: GET {url}: {error_message}")
else: # pragma: no cover
# Server errors: log as error
logger.error(f"Server error: GET {url}: {error_message}")
# Raise a tool error with the friendly message
response.raise_for_status() # Will always raise since we're in the error case
return response # This line will never execute, but it satisfies the type checker # pragma: no cover
except HTTPStatusError as e:
logger.exception(f"Error calling GET {url}: {e}")
raise ToolError(f"Error calling tool: {e}.") from e
status_code = e.response.status_code
error_message = get_error_message(status_code, url, "GET")
raise ToolError(error_message) from e
async def call_put(
@@ -64,6 +159,30 @@ async def call_put(
timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
extensions: RequestExtensions | None = None,
) -> Response:
"""Make a PUT request and handle errors appropriately.
Args:
client: The HTTPX AsyncClient to use
url: The URL to request
content: Request content
data: Form data
files: Files to upload
json: JSON data
params: Query parameters
headers: HTTP headers
cookies: HTTP cookies
auth: Authentication
follow_redirects: Whether to follow redirects
timeout: Request timeout
extensions: HTTPX extensions
Returns:
The HTTP response
Raises:
ToolError: If the request fails with an appropriate error message
"""
logger.debug(f"Calling PUT '{url}'")
try:
response = await client.put(
url,
@@ -79,12 +198,33 @@ async def call_put(
timeout=timeout,
extensions=extensions,
)
logger.debug(response)
response.raise_for_status()
return response
if response.is_success:
return response
# Handle different status codes differently
status_code = response.status_code
error_message = get_error_message(status_code, url, "PUT")
# Log at appropriate level based on status code
if 400 <= status_code < 500:
# Client errors: log as info except for 429 (Too Many Requests)
if status_code == 429: # pragma: no cover
logger.warning(f"Rate limit exceeded: PUT {url}: {error_message}")
else:
logger.info(f"Client error: PUT {url}: {error_message}")
else: # pragma: no cover
# Server errors: log as error
logger.error(f"Server error: PUT {url}: {error_message}")
# Raise a tool error with the friendly message
response.raise_for_status() # Will always raise since we're in the error case
return response # This line will never execute, but it satisfies the type checker # pragma: no cover
except HTTPStatusError as e:
logger.error(f"Error calling PUT {url}: {e}")
raise ToolError(f"Error calling tool: {e}") from e
status_code = e.response.status_code
error_message = get_error_message(status_code, url, "PUT")
raise ToolError(error_message) from e
async def call_post(
@@ -103,6 +243,30 @@ async def call_post(
timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
extensions: RequestExtensions | None = None,
) -> Response:
"""Make a POST request and handle errors appropriately.
Args:
client: The HTTPX AsyncClient to use
url: The URL to request
content: Request content
data: Form data
files: Files to upload
json: JSON data
params: Query parameters
headers: HTTP headers
cookies: HTTP cookies
auth: Authentication
follow_redirects: Whether to follow redirects
timeout: Request timeout
extensions: HTTPX extensions
Returns:
The HTTP response
Raises:
ToolError: If the request fails with an appropriate error message
"""
logger.debug(f"Calling POST '{url}'")
try:
response = await client.post(
url=url,
@@ -118,11 +282,33 @@ async def call_post(
timeout=timeout,
extensions=extensions,
)
response.raise_for_status()
return response
if response.is_success:
return response
# Handle different status codes differently
status_code = response.status_code
error_message = get_error_message(status_code, url, "POST")
# Log at appropriate level based on status code
if 400 <= status_code < 500:
# Client errors: log as info except for 429 (Too Many Requests)
if status_code == 429: # pragma: no cover
logger.warning(f"Rate limit exceeded: POST {url}: {error_message}")
else: # pragma: no cover
logger.info(f"Client error: POST {url}: {error_message}")
else:
# Server errors: log as error
logger.error(f"Server error: POST {url}: {error_message}")
# Raise a tool error with the friendly message
response.raise_for_status() # Will always raise since we're in the error case
return response # This line will never execute, but it satisfies the type checker # pragma: no cover
except HTTPStatusError as e:
logger.error(f"Error calling POST {url}: {e}")
raise ToolError(f"Error calling tool: {e}") from e
status_code = e.response.status_code
error_message = get_error_message(status_code, url, "POST")
raise ToolError(error_message) from e
async def call_delete(
@@ -137,6 +323,26 @@ async def call_delete(
timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
extensions: RequestExtensions | None = None,
) -> Response:
"""Make a DELETE request and handle errors appropriately.
Args:
client: The HTTPX AsyncClient to use
url: The URL to request
params: Query parameters
headers: HTTP headers
cookies: HTTP cookies
auth: Authentication
follow_redirects: Whether to follow redirects
timeout: Request timeout
extensions: HTTPX extensions
Returns:
The HTTP response
Raises:
ToolError: If the request fails with an appropriate error message
"""
logger.debug(f"Calling DELETE '{url}'")
try:
response = await client.delete(
url=url,
@@ -148,8 +354,30 @@ async def call_delete(
timeout=timeout,
extensions=extensions,
)
response.raise_for_status()
return response
if response.is_success:
return response
# Handle different status codes differently
status_code = response.status_code
error_message = get_error_message(status_code, url, "DELETE")
# Log at appropriate level based on status code
if 400 <= status_code < 500:
# Client errors: log as info except for 429 (Too Many Requests)
if status_code == 429: # pragma: no cover
logger.warning(f"Rate limit exceeded: DELETE {url}: {error_message}")
else:
logger.info(f"Client error: DELETE {url}: {error_message}")
else: # pragma: no cover
# Server errors: log as error
logger.error(f"Server error: DELETE {url}: {error_message}")
# Raise a tool error with the friendly message
response.raise_for_status() # Will always raise since we're in the error case
return response # This line will never execute, but it satisfies the type checker # pragma: no cover
except HTTPStatusError as e:
logger.error(f"Error calling DELETE {url}: {e}")
raise ToolError(f"Error calling tool: {e}") from e
status_code = e.response.status_code
error_message = get_error_message(status_code, url, "DELETE")
raise ToolError(error_message) from e
src/basic_memory/mcp/tools/notes.py → src/basic_memory/mcp/tools/write_note.py
+4 -96
View File
@@ -1,20 +1,15 @@
"""Note management tools for Basic Memory MCP server.
These tools provide a natural interface for working with markdown notes
while leveraging the underlying knowledge graph structure.
"""
"""Write note tool for Basic Memory MCP server."""
from typing import Optional, List
from loguru import logger
import logfire
from loguru import logger
from basic_memory.mcp.server import mcp
from basic_memory.mcp.async_client import client
from basic_memory.schemas import EntityResponse, DeleteEntitiesResponse
from basic_memory.schemas.base import Entity
from basic_memory.mcp.tools.utils import call_get, call_put, call_delete
from basic_memory.schemas.memory import memory_url_path
from basic_memory.schemas import EntityResponse
from basic_memory.mcp.tools.utils import call_put
@mcp.tool(
@@ -112,90 +107,3 @@ async def write_note(
summary.append(f"\n## Tags\n- {', '.join(tags)}")
return "\n".join(summary)
@mcp.tool(description="Read note content by title, permalink, relation, or pattern")
async def read_note(identifier: str, page: int = 1, page_size: int = 10) -> str:
"""Get note content in unified diff format.
The content is returned in a unified diff inspired format:
```
--- memory://docs/example 2025-01-31T19:32:49 7d9f1c8b
<document content>
```
Multiple documents (from relations or pattern matches) are separated by
additional headers.
Args:
identifier: Can be one of:
- Note title ("Project Planning")
- Note permalink ("docs/example")
- Relation path ("docs/example/depends-on/other-doc")
- Pattern match ("docs/*-architecture")
page: the page number of results to return (default 1)
page_size: the number of results to return per page (default 10)
Returns:
Document content in unified diff format. For single documents, returns
just that document's content. For relations or pattern matches, returns
multiple documents separated by unified diff headers.
Examples:
# Single document
content = await read_note("Project Planning")
# Read by permalink
content = await read_note("docs/architecture/file-first")
# Follow relation
content = await read_note("docs/architecture/depends-on/docs/content-parser")
# Pattern matching
content = await read_note("docs/*-architecture") # All architecture docs
content = await read_note("docs/*/implements/*") # Find implementations
Output format:
```
--- memory://docs/example 2025-01-31T19:32:49 7d9f1c8b
<first document content>
--- memory://docs/other 2025-01-30T15:45:22 a1b2c3d4
<second document content>
```
The headers include:
- Full memory:// URI for the document
- Last modified timestamp
- Content checksum
"""
with logfire.span("Reading note", identifier=identifier): # pyright: ignore [reportGeneralTypeIssues]
logger.info(f"Reading note {identifier}")
url = memory_url_path(identifier)
response = await call_get(
client, f"/resource/{url}", params={"page": page, "page_size": page_size}
)
return response.text
@mcp.tool(description="Delete a note by title or permalink")
async def delete_note(identifier: str) -> bool:
"""Delete a note from the knowledge base.
Args:
identifier: Note title or permalink
Returns:
True if note was deleted, False otherwise
Examples:
# Delete by title
delete_note("Meeting Notes: Project Planning")
# Delete by permalink
delete_note("notes/project-planning")
"""
with logfire.span("Deleting note", identifier=identifier): # pyright: ignore [reportGeneralTypeIssues]
response = await call_delete(client, f"/knowledge/entities/{identifier}")
result = DeleteEntitiesResponse.model_validate(response.json())
return result.deleted
src/basic_memory/models/search.py
+2 -1
View File
@@ -8,7 +8,8 @@ CREATE VIRTUAL TABLE IF NOT EXISTS search_index USING fts5(
-- Core entity fields
id UNINDEXED, -- Row ID
title, -- Title for searching
content, -- Main searchable content
content_stems, -- Main searchable content split into stems
content_snippet, -- File content snippet for display
permalink, -- Stable identifier (now indexed for path search)
file_path UNINDEXED, -- Physical location
type UNINDEXED, -- entity/relation/observation
src/basic_memory/repository/entity_repository.py
+3 -2
View File
@@ -31,14 +31,15 @@ class EntityRepository(Repository[Entity]):
query = self.select().where(Entity.permalink == permalink).options(*self.get_load_options())
return await self.find_one(query)
async def get_by_title(self, title: str) -> Optional[Entity]:
async def get_by_title(self, title: str) -> Sequence[Entity]:
"""Get entity by title.
Args:
title: Title of the entity to find
"""
query = self.select().where(Entity.title == title).options(*self.get_load_options())
return await self.find_one(query)
result = await self.execute_query(query)
return list(result.scalars().all())
async def get_by_file_path(self, file_path: Union[Path, str]) -> Optional[Entity]:
"""Get entity by file_path.
src/basic_memory/repository/search_repository.py
+13 -7
View File
@@ -35,18 +35,24 @@ class SearchIndexRow:
# Type-specific fields
title: Optional[str] = None # entity
content: Optional[str] = None # entity, observation
content_stems: Optional[str] = None # entity, observation
content_snippet: Optional[str] = None # entity, observation
entity_id: Optional[int] = None # observations
category: Optional[str] = None # observations
from_id: Optional[int] = None # relations
to_id: Optional[int] = None # relations
relation_type: Optional[str] = None # relations
@property
def content(self):
return self.content_snippet
def to_insert(self):
return {
"id": self.id,
"title": self.title,
"content": self.content,
"content_stems": self.content_stems,
"content_snippet": self.content_snippet,
"permalink": self.permalink,
"file_path": self.file_path,
"type": self.type,
@@ -126,7 +132,7 @@ class SearchRepository:
if search_text:
search_text = self._prepare_search_term(search_text.strip())
params["text"] = search_text
conditions.append("(title MATCH :text OR content MATCH :text)")
conditions.append("(title MATCH :text OR content_stems MATCH :text)")
# Handle title match search
if title:
@@ -188,7 +194,7 @@ class SearchRepository:
to_id,
relation_type,
entity_id,
content,
content_snippet,
category,
created_at,
updated_at,
@@ -218,7 +224,7 @@ class SearchRepository:
to_id=row.to_id,
relation_type=row.relation_type,
entity_id=row.entity_id,
content=row.content,
content_snippet=row.content_snippet,
category=row.category,
created_at=row.created_at,
updated_at=row.updated_at,
@@ -250,12 +256,12 @@ class SearchRepository:
await session.execute(
text("""
INSERT INTO search_index (
id, title, content, permalink, file_path, type, metadata,
id, title, content_stems, content_snippet, permalink, file_path, type, metadata,
from_id, to_id, relation_type,
entity_id, category,
created_at, updated_at
) VALUES (
:id, :title, :content, :permalink, :file_path, :type, :metadata,
:id, :title, :content_stems, :content_snippet, :permalink, :file_path, :type, :metadata,
:from_id, :to_id, :relation_type,
:entity_id, :category,
:created_at, :updated_at
src/basic_memory/schemas/memory.py
+3 -2
View File
@@ -64,6 +64,7 @@ class EntitySummary(BaseModel):
type: str = "entity"
permalink: Optional[str]
title: str
content: Optional[str] = None
file_path: str
created_at: datetime
@@ -76,8 +77,8 @@ class RelationSummary(BaseModel):
file_path: str
permalink: str
relation_type: str
from_id: str
to_id: Optional[str] = None
from_entity: str
to_entity: Optional[str] = None
created_at: datetime
src/basic_memory/schemas/search.py
+7 -20
View File
@@ -11,6 +11,8 @@ from datetime import datetime
from enum import Enum
from pydantic import BaseModel, field_validator
from basic_memory.schemas.base import Permalink
class SearchItemType(str, Enum):
"""Types of searchable items."""
@@ -36,7 +38,7 @@ class SearchQuery(BaseModel):
# Primary search modes (use ONE of these)
permalink: Optional[str] = None # Exact permalink match
permalink_match: Optional[str] = None # Exact permalink match
permalink_match: Optional[str] = None # Glob permalink match
text: Optional[str] = None # Full-text search
title: Optional[str] = None # title only search
@@ -67,38 +69,23 @@ class SearchQuery(BaseModel):
class SearchResult(BaseModel):
"""Search result with score and metadata."""
id: int
title: str
type: SearchItemType
score: float
entity: Optional[Permalink]
permalink: Optional[str]
content: Optional[str] = None
file_path: str
metadata: Optional[dict] = None
# Type-specific fields
entity_id: Optional[int] = None # For observations
category: Optional[str] = None # For observations
from_id: Optional[int] = None # For relations
to_id: Optional[int] = None # For relations
from_entity: Optional[Permalink] = None # For relations
to_entity: Optional[Permalink] = None # For relations
relation_type: Optional[str] = None # For relations
class RelatedResult(BaseModel):
type: SearchItemType
id: int
title: str
permalink: str
depth: int
root_id: int
created_at: datetime
from_id: Optional[int] = None
to_id: Optional[int] = None
relation_type: Optional[str] = None
category: Optional[str] = None
entity_id: Optional[int] = None
class SearchResponse(BaseModel):
"""Wrapper for search results."""
src/basic_memory/services/context_service.py
+3 -3
View File
@@ -165,7 +165,7 @@ class ContextService:
from_id,
to_id,
relation_type,
content,
content_snippet as content,
category,
entity_id,
0 as depth,
@@ -189,7 +189,7 @@ class ContextService:
r.from_id,
r.to_id,
r.relation_type,
r.content,
r.content_snippet as content,
r.category,
r.entity_id,
cg.depth + 1,
@@ -218,7 +218,7 @@ class ContextService:
e.from_id,
e.to_id,
e.relation_type,
e.content,
e.content_snippet as content,
e.category,
e.entity_id,
cg.depth + 1, -- Increment depth for entities
src/basic_memory/services/link_resolver.py
+5 -45
View File
@@ -1,12 +1,11 @@
"""Service for resolving markdown links to permalinks."""
from typing import Optional, Tuple, List
from typing import Optional, Tuple
from loguru import logger
from basic_memory.models import Entity
from basic_memory.repository.entity_repository import EntityRepository
from basic_memory.repository.search_repository import SearchIndexRow
from basic_memory.schemas.search import SearchQuery, SearchItemType
from basic_memory.services.search_service import SearchService
@@ -41,8 +40,9 @@ class LinkResolver:
return entity
# 2. Try exact title match
entity = await self.entity_repository.get_by_title(clean_text)
if entity:
found = await self.entity_repository.get_by_title(clean_text)
if found and len(found) == 1:
entity = found[0]
logger.debug(f"Found title match: {entity.title}")
return entity
@@ -54,7 +54,7 @@ class LinkResolver:
if results:
# Look for best match
best_match = self._select_best_match(clean_text, results)
best_match = min(results, key=lambda x: x.score) # pyright: ignore
logger.debug(
f"Selected best match from {len(results)} results: {best_match.permalink}"
)
@@ -88,43 +88,3 @@ class LinkResolver:
alias = alias.strip()
return text, alias
def _select_best_match(self, search_text: str, results: List[SearchIndexRow]) -> SearchIndexRow:
"""Select best match from search results.
Uses multiple criteria:
1. Word matches in title field
2. Word matches in path
3. Overall search score
"""
# Get search terms for matching
terms = search_text.lower().split()
# Score each result
scored_results = []
for result in results:
# Start with base score (lower is better)
score = result.score or 0
if result.permalink:
# Parse path components
path_parts = result.permalink.lower().split("/")
last_part = path_parts[-1] if path_parts else ""
else:
last_part = "" # pragma: no cover
# Title word match boosts
term_matches = [term for term in terms if term in last_part]
if term_matches:
score *= 0.5 # Boost for each matching term
# Exact title match is best
if last_part == search_text.lower():
score *= 0.2
scored_results.append((score, result))
# Sort by score (lowest first) and return best
scored_results.sort(key=lambda x: x[0], reverse=True)
return scored_results[0][1]
src/basic_memory/services/search_service.py
+22 -9
View File
@@ -145,6 +145,7 @@ class SearchService:
await self.repository.index_item(
SearchIndexRow(
id=entity.id,
entity_id=entity.id,
type=SearchItemType.ENTITY.value,
title=entity.title,
file_path=entity.file_path,
@@ -182,29 +183,33 @@ class SearchService:
"entity.permalink should not be None for markdown entities"
)
content_parts = []
content_stems = []
content_snippet = ""
title_variants = self._generate_variants(entity.title)
content_parts.extend(title_variants)
content_stems.extend(title_variants)
content = await self.file_service.read_entity_content(entity)
if content:
content_parts.append(content)
content_stems.append(content)
content_snippet = f"{content[:250]}"
content_parts.extend(self._generate_variants(entity.permalink))
content_parts.extend(self._generate_variants(entity.file_path))
content_stems.extend(self._generate_variants(entity.permalink))
content_stems.extend(self._generate_variants(entity.file_path))
entity_content = "\n".join(p for p in content_parts if p and p.strip())
entity_content_stems = "\n".join(p for p in content_stems if p and p.strip())
assert entity.permalink is not None, (
"entity.permalink should not be None for markdown entities"
)
# Index entity
await self.repository.index_item(
SearchIndexRow(
id=entity.id,
type=SearchItemType.ENTITY.value,
title=entity.title,
content=entity_content,
content_stems=entity_content_stems,
content_snippet=content_snippet,
permalink=entity.permalink,
file_path=entity.file_path,
entity_id=entity.id,
@@ -219,12 +224,16 @@ class SearchService:
# Index each observation with permalink
for obs in entity.observations:
# Index with parent entity's file path since that's where it's defined
obs_content_stems = "\n".join(
p for p in self._generate_variants(obs.content) if p and p.strip()
)
await self.repository.index_item(
SearchIndexRow(
id=obs.id,
type=SearchItemType.OBSERVATION.value,
title=f"{obs.category}: {obs.content[:50]}...",
content=obs.content,
title=f"{obs.category}: {obs.content[:100]}...",
content_stems=obs_content_stems,
content_snippet=obs.content,
permalink=obs.permalink,
file_path=entity.file_path,
category=obs.category,
@@ -246,11 +255,15 @@ class SearchService:
else f"{rel.from_entity.title}"
)
rel_content_stems = "\n".join(
p for p in self._generate_variants(relation_title) if p and p.strip()
)
await self.repository.index_item(
SearchIndexRow(
id=rel.id,
title=relation_title,
permalink=rel.permalink,
content_stems=rel_content_stems,
file_path=entity.file_path,
type=SearchItemType.RELATION.value,
entity_id=entity.id,
src/basic_memory/sync/sync_service.py
+14 -10
View File
@@ -10,6 +10,7 @@ from typing import Tuple
import logfire
from loguru import logger
from sqlalchemy.exc import IntegrityError
from basic_memory.markdown import EntityParser
from basic_memory.models import Entity
@@ -178,11 +179,11 @@ class SyncService:
return entity, checksum
except Exception as e: # pragma: no cover
logger.error(f"Failed to sync {path}: {e}")
raise
logger.exception(f"Failed to sync {path}: {e}")
return None, None # pyright: ignore
async def sync_markdown_file(self, path: str, new: bool = True) -> Tuple[Entity, str]:
"""Sync a markdown file with full processing."""
"""Sync a markdown file with full proces sing."""
# Parse markdown first to get any existing permalink
entity_markdown = await self.entity_parser.parse_file(path)
@@ -301,13 +302,16 @@ class SyncService:
logger.debug(
f"Resolved forward reference: {relation.to_name} -> {resolved_entity.title}"
)
await self.relation_repository.update(
relation.id,
{
"to_id": resolved_entity.id,
"to_name": resolved_entity.title,
},
)
try:
await self.relation_repository.update(
relation.id,
{
"to_id": resolved_entity.id,
"to_name": resolved_entity.title,
},
)
except IntegrityError: # pragma: no cover
logger.debug(f"Ignoring duplicate relation {relation}")
# update search index
await self.search_service.index_entity(resolved_entity)
src/basic_memory/sync/watch_service.py
+9 -4
View File
@@ -192,9 +192,14 @@ class WatchService:
for path in adds:
if path not in processed:
_, checksum = await self.sync_service.sync_file(path, new=True)
self.state.add_event(path=path, action="new", status="success", checksum=checksum)
self.console.print(f"[green]✓[/green] Added: {path}")
processed.add(path)
if checksum:
self.state.add_event(
path=path, action="new", status="success", checksum=checksum
)
self.console.print(f"[green]✓[/green] Added: {path}")
processed.add(path)
else:
self.console.print(f"[orange]?[/orange] Error syncing: {path}")
for path in modifies:
if path not in processed:
@@ -207,7 +212,7 @@ class WatchService:
# Add a divider if we processed any files
if processed:
self.console.print("─" * 50, style="dim")
self.console.print("─" * 80, style="dim")
self.state.last_scan = datetime.now()
self.state.synced_files += len(processed)
src/basic_memory/utils.py
+33 -15
View File
@@ -14,6 +14,11 @@ import basic_memory
import logfire
# Disable the "Queue is full" warning
logging.getLogger("opentelemetry.sdk.metrics._internal.instrument").setLevel(logging.ERROR)
# Disable logfire prompts in CI/automated environments
os.environ["LOGFIRE_IGNORE_NO_CONFIG"] = "1"
def generate_permalink(file_path: Union[Path, str]) -> str:
"""Generate a stable permalink from a file path.
@@ -84,20 +89,26 @@ def setup_logging(
# Add file handler if we are not running tests
if log_file and env != "test":
# enable pydantic logfire
logfire.configure(
code_source=logfire.CodeSource(
repository="https://github.com/basicmachines-co/basic-memory",
revision=basic_memory.__version__,
),
environment=env,
console=False,
)
logger.configure(handlers=[logfire.loguru_handler()])
try:
# Skip logfire configuration if LOGFIRE_API_KEY is not set
# This avoids interactive prompts when running automated tasks
if "LOGFIRE_API_KEY" in os.environ:
# enable pydantic logfire
logfire.configure(
code_source=logfire.CodeSource(
repository="https://github.com/basicmachines-co/basic-memory",
revision=basic_memory.__version__,
),
environment=env,
console=False,
)
logger.configure(handlers=[logfire.loguru_handler()])
# instrument code spans
logfire.instrument_sqlite3()
logfire.instrument_httpx()
# instrument code spans
logfire.instrument_sqlite3()
logfire.instrument_httpx()
except Exception as e:
logger.warning(f"Failed to configure logfire: {e}")
# setup logger
log_path = home_dir / log_file
@@ -126,5 +137,12 @@ def setup_logging(
# turn watchfiles to WARNING
logging.getLogger("watchfiles.main").setLevel(logging.WARNING)
# disable open telemetry warning
logging.getLogger("instrumentor").setLevel(logging.ERROR)
# Disable all instrumentor-related warnings
for logger_name in [
"instrumentor",
"opentelemetry.instrumentation.instrumentor",
"opentelemetry.instrumentation",
"logfire.instrumentor",
"opentelemetry.sdk.metrics._internal.instrument",
]:
logging.getLogger(logger_name).setLevel(logging.ERROR)