recent_activity
Retrieve recent knowledge base activity using natural language timeframes like 'yesterday' or 'last week' to track updates and changes.
Instructions
Get recent activity from across the knowledge base.
Timeframe supports natural language formats like:
- "2 days ago"
- "last week"
- "yesterday"
- "today"
- "3 weeks ago"
Or standard formats like "7d"
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| depth | No | ||
| max_related | No | ||
| page | No | ||
| page_size | No | ||
| project | No | ||
| timeframe | No | 7d | |
| type | No |
Implementation Reference
- The core handler function `recent_activity` decorated with `@mcp.tool()`, implementing the full logic for fetching, processing, and formatting recent activity data from projects or across all projects. Handles project resolution, type filtering, timeframes, discovery mode, and formats human-readable output.@mcp.tool( description="""Get recent activity for a project or across all projects. Timeframe supports natural language formats like: - "2 days ago" - "last week" - "yesterday" - "today" - "3 weeks ago" Or standard formats like "7d" """, ) async def recent_activity( type: Union[str, List[str]] = "", depth: int = 1, timeframe: TimeFrame = "7d", project: Optional[str] = None, context: Context | None = None, ) -> str: """Get recent activity for a specific project or across all projects. Project Resolution: The server resolves projects in this order: 1. Single Project Mode - server constrained to one project, parameter ignored 2. Explicit project parameter - specify which project to query 3. Default project - server configured default if no project specified Discovery Mode: When no specific project can be resolved, returns activity across all projects to help discover available projects and their recent activity. Project Discovery (when project is unknown): 1. Call list_memory_projects() to see available projects 2. Or use this tool without project parameter to see cross-project activity 3. Ask the user which project to focus on 4. Remember their choice for the conversation Args: 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" - Points in time: "2024-01-01", "January 1st" - Standard format: "7d", "24h" project: Project name to query. Optional - server will resolve using the hierarchy above. If unknown, use list_memory_projects() to discover available projects. context: Optional FastMCP context for performance caching. Returns: Human-readable summary of recent activity. When no specific project is resolved, returns cross-project discovery information. When a specific project is resolved, returns detailed activity for that project. Examples: # Cross-project discovery mode recent_activity() recent_activity(timeframe="yesterday") # Project-specific activity recent_activity(project="work-docs", type="entity", timeframe="yesterday") recent_activity(project="research", type=["entity", "relation"], timeframe="today") recent_activity(project="notes", type="entity", depth=2, timeframe="2 weeks ago") Raises: ToolError: If project doesn't exist or type parameter contains invalid values Notes: - Higher depth values (>3) may impact performance with large result sets - For focused queries, consider using build_context with a specific URI - Max timeframe is 1 year in the past """ async with get_client() as client: # Build common parameters for API calls params = { "page": 1, "page_size": 10, "max_related": 10, } if depth: params["depth"] = depth if timeframe: params["timeframe"] = timeframe # pyright: ignore # Validate and convert type parameter if 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 # Resolve project parameter using the three-tier hierarchy resolved_project = await resolve_project_parameter(project) if resolved_project is None: # Discovery Mode: Get activity across all projects logger.info( f"Getting recent activity across all projects: type={type}, depth={depth}, timeframe={timeframe}" ) # Get list of all projects response = await call_get(client, "/projects/projects") project_list = ProjectList.model_validate(response.json()) projects_activity = {} total_items = 0 total_entities = 0 total_relations = 0 total_observations = 0 most_active_project = None most_active_count = 0 active_projects = 0 # Query each project's activity for project_info in project_list.projects: project_activity = await _get_project_activity(client, project_info, params, depth) projects_activity[project_info.name] = project_activity # Aggregate stats item_count = project_activity.item_count if item_count > 0: active_projects += 1 total_items += item_count # Count by type for result in project_activity.activity.results: if result.primary_result.type == "entity": total_entities += 1 elif result.primary_result.type == "relation": total_relations += 1 elif result.primary_result.type == "observation": total_observations += 1 # Track most active project if item_count > most_active_count: most_active_count = item_count most_active_project = project_info.name # Build summary stats summary = ActivityStats( total_projects=len(project_list.projects), active_projects=active_projects, most_active_project=most_active_project, total_items=total_items, total_entities=total_entities, total_relations=total_relations, total_observations=total_observations, ) # Generate guidance for the assistant guidance_lines = ["\n" + "─" * 40] if most_active_project and most_active_count > 0: guidance_lines.extend( [ f"Suggested project: '{most_active_project}' (most active with {most_active_count} items)", f"Ask user: 'Should I use {most_active_project} for this task, or would you prefer a different project?'", ] ) elif active_projects > 0: # Has activity but no clear most active project active_project_names = [ name for name, activity in projects_activity.items() if activity.item_count > 0 ] if len(active_project_names) == 1: guidance_lines.extend( [ f"Suggested project: '{active_project_names[0]}' (only active project)", f"Ask user: 'Should I use {active_project_names[0]} for this task?'", ] ) else: guidance_lines.extend( [ f"Multiple active projects found: {', '.join(active_project_names)}", "Ask user: 'Which project should I use for this task?'", ] ) else: # No recent activity guidance_lines.extend( [ "No recent activity found in any project.", "Consider: Ask which project to use or if they want to create a new one.", ] ) guidance_lines.extend( [ "", "Session reminder: Remember their project choice throughout this conversation.", ] ) guidance = "\n".join(guidance_lines) # Format discovery mode output return _format_discovery_output(projects_activity, summary, timeframe, guidance) else: # Project-Specific Mode: Get activity for specific project logger.info( f"Getting recent activity from project {resolved_project}: type={type}, depth={depth}, timeframe={timeframe}" ) active_project = await get_active_project(client, resolved_project, context) project_url = active_project.project_url response = await call_get( client, f"{project_url}/memory/recent", params=params, ) activity_data = GraphContext.model_validate(response.json()) # Format project-specific mode output return _format_project_output(resolved_project, activity_data, timeframe, type)
- Helper function to fetch activity data for a single project during discovery mode.async def _get_project_activity( client, project_info: ProjectItem, params: dict, depth: int ) -> ProjectActivity: """Get activity data for a single project. Args: client: HTTP client for API calls project_info: Project information params: Query parameters for the activity request depth: Graph traversal depth Returns: ProjectActivity with activity data or empty activity on error """ project_url = f"/{project_info.permalink}" activity_response = await call_get( client, f"{project_url}/memory/recent", params=params, ) activity = GraphContext.model_validate(activity_response.json()) # Extract last activity timestamp and active folders last_activity = None active_folders = set() for result in activity.results: if result.primary_result.created_at: current_time = result.primary_result.created_at try: if last_activity is None or current_time > last_activity: last_activity = current_time except TypeError: # Handle timezone comparison issues by skipping this comparison if last_activity is None: last_activity = current_time # Extract folder from file_path if hasattr(result.primary_result, "file_path") and result.primary_result.file_path: folder = "/".join(result.primary_result.file_path.split("/")[:-1]) if folder: active_folders.add(folder) return ProjectActivity( project_name=project_info.name, project_path=project_info.path, activity=activity, item_count=len(activity.results), last_activity=last_activity, active_folders=list(active_folders)[:5], # Limit to top 5 folders )
- Calls to formatting helpers like _format_project_output and _format_discovery_output for output generation.return _format_project_output(resolved_project, activity_data, timeframe, type)
- src/basic_memory/mcp/tools/recent_activity.py:22-33 (registration)The @mcp.tool decorator registers the recent_activity tool with MCP server.@mcp.tool( description="""Get recent activity for a project or across all projects. Timeframe supports natural language formats like: - "2 days ago" - "last week" - "yesterday" - "today" - "3 weeks ago" Or standard formats like "7d" """, )