basic-memory

Instructions

Timeframe supports natural language formats like:
- "2 days ago"
- "last week"
- "yesterday"
- "today"
- "3 weeks ago"
Or standard formats like "7d"

Implementation Reference

• src/basic_memory/mcp/tools/recent_activity.py:24-255 (handler)

  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