recent_activity
Retrieve recent project activity from your knowledge management system using natural language timeframes like 'yesterday' or 'last week'.
Instructions
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"
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| type | No | ||
| depth | No | ||
| timeframe | No | 7d | |
| project | No |
Implementation Reference
- The primary handler function for the 'recent_activity' MCP tool. It resolves the project context, fetches recent graph activity via API calls, handles discovery mode across projects or specific project mode, and formats the results into a human-readable markdown summary with stats, key items, and guidance.@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 """ track_mcp_tool("recent_activity") 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 # allow_discovery=True enables Discovery Mode, so a project is not required resolved_project = await resolve_project_parameter(project, allow_discovery=True) 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 active_projects == 0: # 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.", ] ) else: # At least one project has activity: suggest the most active project. suggested_project = most_active_project or next( (name for name, activity in projects_activity.items() if activity.item_count > 0), None, ) if suggested_project: suffix = ( f"(most active with {most_active_count} items)" if most_active_count > 0 else "" ) guidance_lines.append(f"Suggested project: '{suggested_project}' {suffix}".strip()) if active_projects == 1: guidance_lines.append(f"Ask user: 'Should I use {suggested_project} for this task?'") else: guidance_lines.append( f"Ask user: 'Should I use {suggested_project} for this task, or would you prefer a different project?'" ) 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) response = await call_get( client, f"/v2/projects/{active_project.external_id}/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)
- src/basic_memory/mcp/tools/__init__.py:12-12 (registration)The import statement in tools/__init__.py that registers the recent_activity tool with the FastMCP server instance by executing the module which defines @mcp.tool()-decorated function.from basic_memory.mcp.tools.recent_activity import recent_activity