fix: modify recent_activity args to be strings instead of enums

Signed-off-by: phernandez <paul@basicmachines.co>
This commit is contained in:
phernandez
2025-04-17 10:48:35 -05:00
committed by Paul Hernandez
parent 81616ab42e
commit 3c1cc346df
3 changed files with 148 additions and 66 deletions
+37 -15
View File
@@ -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
@@ -14,7 +14,7 @@ from basic_memory.schemas.search import SearchItemType
@mcp.tool(
description="""Get recent activity from across the knowledge base.
Timeframe supports natural language formats like:
- "2 days ago"
- "last week"
@@ -25,9 +25,9 @@ from basic_memory.schemas.search import SearchItemType
""",
)
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,
@@ -35,11 +35,14 @@ async def recent_activity(
"""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"
@@ -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
@@ -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,
@@ -5,12 +5,9 @@ from datetime import datetime
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,
)
@@ -83,53 +80,6 @@ invalid_timeframes = [
]
@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."""
+110
View File
@@ -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']"
)