Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 15 additions & 1 deletion CLAUDE.md
Original file line number Diff line number Diff line change
Expand Up @@ -172,4 +172,18 @@ With GitHub integration, the development workflow includes:
4. **Documentation maintenance** - Claude can keep documentation updated as the code evolves

With this integration, the AI assistant is a full-fledged team member rather than just a tool for generating code
snippets.
snippets.


### Basic Memory Pro

Basic Memory Pro is a desktop GUI application that wraps the basic-memory CLI/MCP tools:

- Built with Tauri (Rust), React (TypeScript), and a Python FastAPI sidecar
- Provides visual knowledge graph exploration and project management
- Uses the same core codebase but adds a desktop-friendly interface
- Project configuration is shared between CLI and Pro versions
- Multiple project support with visual switching interface

local repo: /Users/phernandez/dev/basicmachines/basic-memory-pro
github: https://github.com/basicmachines-co/basic-memory-pro
52 changes: 37 additions & 15 deletions src/basic_memory/mcp/tools/recent_activity.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
"""Recent activity tool for Basic Memory MCP server."""

from typing import Optional, List
from typing import List, Union

from loguru import logger

Expand All @@ -14,7 +14,7 @@

@mcp.tool(
description="""Get recent activity from across the knowledge base.

Timeframe supports natural language formats like:
- "2 days ago"
- "last week"
Expand All @@ -25,21 +25,24 @@
""",
)
async def recent_activity(
type: Optional[List[SearchItemType]] = None,
depth: Optional[int] = 1,
timeframe: Optional[TimeFrame] = "7d",
type: Union[str, List[str]] = "",
depth: int = 1,
timeframe: TimeFrame = "7d",
page: int = 1,
page_size: int = 10,
max_related: int = 10,
) -> GraphContext:
"""Get recent activity across the knowledge base.

Args:
type: Filter by content type(s). Valid options:
- ["entity"] for knowledge entities
- ["relation"] for connections between entities
- ["observation"] for notes and observations
type: Filter by content type(s). Can be a string or list of strings.
Valid options:
- "entity" or ["entity"] for knowledge entities
- "relation" or ["relation"] for connections between entities
- "observation" or ["observation"] for notes and observations
Multiple types can be combined: ["entity", "relation"]
Case-insensitive: "ENTITY" and "entity" are treated the same.
Default is an empty string, which returns all types.
depth: How many relation hops to traverse (1-3 recommended)
timeframe: Time window to search. Supports natural language:
- Relative: "2 days ago", "last week", "yesterday"
Expand All @@ -59,14 +62,17 @@ async def recent_activity(
# Get all entities for the last 10 days (default)
recent_activity()

# Get all entities from yesterday
# Get all entities from yesterday (string format)
recent_activity(type="entity", timeframe="yesterday")

# Get all entities from yesterday (list format)
recent_activity(type=["entity"], timeframe="yesterday")

# Get recent relations and observations
recent_activity(type=["relation", "observation"], timeframe="today")

# Look back further with more context
recent_activity(type=["entity"], depth=2, timeframe="2 weeks ago")
recent_activity(type="entity", depth=2, timeframe="2 weeks ago")

Notes:
- Higher depth values (>3) may impact performance with large result sets
Expand All @@ -86,11 +92,27 @@ async def recent_activity(
if timeframe:
params["timeframe"] = timeframe # pyright: ignore

# send enum values if we have an enum, else send string value
# Validate and convert type parameter
if type:
params["type"] = [ # pyright: ignore
type.value if isinstance(type, SearchItemType) else type for type in type
]
# Convert single string to list
if isinstance(type, str):
type_list = [type]
else:
type_list = type

# Validate each type against SearchItemType enum
validated_types = []
for t in type_list:
try:
# Try to convert string to enum
if isinstance(t, str):
validated_types.append(SearchItemType(t.lower()))
except ValueError:
valid_types = [t.value for t in SearchItemType]
raise ValueError(f"Invalid type: {t}. Valid types are: {valid_types}")

# Add validated types to params
params["type"] = [t.value for t in validated_types] # pyright: ignore

response = await call_get(
client,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -5,12 +5,9 @@

from mcp.server.fastmcp.exceptions import ToolError

from basic_memory.mcp.tools import build_context, recent_activity
from basic_memory.mcp.tools import build_context
from basic_memory.schemas.memory import (
GraphContext,
EntitySummary,
ObservationSummary,
RelationSummary,
)


Expand Down Expand Up @@ -83,53 +80,6 @@ async def test_get_discussion_context_not_found(client):
]


@pytest.mark.asyncio
async def test_recent_activity_timeframe_formats(client, test_graph):
"""Test that recent_activity accepts various timeframe formats."""
# Test each valid timeframe
for timeframe in valid_timeframes:
try:
result = await recent_activity(
type=["entity"], timeframe=timeframe, page=1, page_size=10, max_related=10
)
assert result is not None
except Exception as e:
pytest.fail(f"Failed with valid timeframe '{timeframe}': {str(e)}")

# Test invalid timeframes should raise ValidationError
for timeframe in invalid_timeframes:
with pytest.raises(ToolError):
await recent_activity(timeframe=timeframe)


@pytest.mark.asyncio
async def test_recent_activity_type_filters(client, test_graph):
"""Test that recent_activity correctly filters by types."""
# Test single type
result = await recent_activity(type=["entity"])
assert result is not None
assert all(isinstance(r, EntitySummary) for r in result.primary_results)

# Test multiple types
result = await recent_activity(type=["entity", "observation"])
assert result is not None
assert all(
isinstance(r, EntitySummary) or isinstance(r, ObservationSummary)
for r in result.primary_results
)

# Test all types
result = await recent_activity(type=["entity", "observation", "relation"])
assert result is not None
# Results can be any type
assert all(
isinstance(r, EntitySummary)
or isinstance(r, ObservationSummary)
or isinstance(r, RelationSummary)
for r in result.primary_results
)


@pytest.mark.asyncio
async def test_build_context_timeframe_formats(client, test_graph):
"""Test that build_context accepts various timeframe formats."""
Expand Down
110 changes: 110 additions & 0 deletions tests/mcp/test_tool_recent_activity.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,110 @@
"""Tests for discussion context MCP tool."""

import pytest

from mcp.server.fastmcp.exceptions import ToolError

from basic_memory.mcp.tools import recent_activity
from basic_memory.schemas.memory import (
EntitySummary,
ObservationSummary,
RelationSummary,
)
from basic_memory.schemas.search import SearchItemType

# Test data for different timeframe formats
valid_timeframes = [
"7d", # Standard format
"yesterday", # Natural language
"0d", # Zero duration
]

invalid_timeframes = [
"invalid", # Nonsense string
"tomorrow", # Future date
]


@pytest.mark.asyncio
async def test_recent_activity_timeframe_formats(client, test_graph):
"""Test that recent_activity accepts various timeframe formats."""
# Test each valid timeframe
for timeframe in valid_timeframes:
try:
result = await recent_activity(
type=["entity"], timeframe=timeframe, page=1, page_size=10, max_related=10
)
assert result is not None
except Exception as e:
pytest.fail(f"Failed with valid timeframe '{timeframe}': {str(e)}")

# Test invalid timeframes should raise ValidationError
for timeframe in invalid_timeframes:
with pytest.raises(ToolError):
await recent_activity(timeframe=timeframe)


@pytest.mark.asyncio
async def test_recent_activity_type_filters(client, test_graph):
"""Test that recent_activity correctly filters by types."""

# Test single string type
result = await recent_activity(type=SearchItemType.ENTITY)
assert result is not None
assert all(isinstance(r, EntitySummary) for r in result.primary_results)

# Test single string type
result = await recent_activity(type="entity")
assert result is not None
assert all(isinstance(r, EntitySummary) for r in result.primary_results)

# Test single type
result = await recent_activity(type=["entity"])
assert result is not None
assert all(isinstance(r, EntitySummary) for r in result.primary_results)

# Test multiple types
result = await recent_activity(type=["entity", "observation"])
assert result is not None
assert all(
isinstance(r, EntitySummary) or isinstance(r, ObservationSummary)
for r in result.primary_results
)

# Test multiple types
result = await recent_activity(type=[SearchItemType.ENTITY, SearchItemType.OBSERVATION])
assert result is not None
assert all(
isinstance(r, EntitySummary) or isinstance(r, ObservationSummary)
for r in result.primary_results
)

# Test all types
result = await recent_activity(type=["entity", "observation", "relation"])
assert result is not None
# Results can be any type
assert all(
isinstance(r, EntitySummary)
or isinstance(r, ObservationSummary)
or isinstance(r, RelationSummary)
for r in result.primary_results
)


@pytest.mark.asyncio
async def test_recent_activity_type_invalid(client, test_graph):
"""Test that recent_activity correctly filters by types."""

# Test single invalid string type
with pytest.raises(ValueError) as e:
await recent_activity(type="note")
assert (
str(e.value) == "Invalid type: note. Valid types are: ['entity', 'observation', 'relation']"
)

# Test invalid string array type
with pytest.raises(ValueError) as e:
await recent_activity(type=["note"])
assert (
str(e.value) == "Invalid type: note. Valid types are: ['entity', 'observation', 'relation']"
)
Loading