get_episodes

Instructions

Get the most recent memory episodes for a specific group.

Args:
    group_id: ID of the group to retrieve episodes from. If not provided, uses the default group_id.
    last_n: Number of most recent episodes to retrieve (default: 10)

Returns:
    List of episode dictionaries

Example:
    get_episodes(group_id="knowledge-smith", last_n=5)

@REQ: REQ-graphiti-chunk-mcp
@BP: BP-graphiti-chunk-mcp
@TASK: TASK-007-MCPTools

Implementation Reference

• rbt_mcp_server/server.py:108-134 (registration)

  MCP tool registration and handler for 'get_episodes'. Decorated with @mcp.tool() and delegates implementation to graphiti_tools.get_episodes_impl.

  @mcp.tool()
  async def get_episodes(
      group_id: Optional[str] = None,
      last_n: int = 10,
  ) -> List[Dict[str, Any]]:
      """
      Get the most recent memory episodes for a specific group.
  
      Args:
          group_id: ID of the group to retrieve episodes from. If not provided, uses the default group_id.
          last_n: Number of most recent episodes to retrieve (default: 10)
  
      Returns:
          List of episode dictionaries
  
      Example:
          get_episodes(group_id="knowledge-smith", last_n=5)
  
      @REQ: REQ-graphiti-chunk-mcp
      @BP: BP-graphiti-chunk-mcp
      @TASK: TASK-007-MCPTools
      """
      return await graphiti_tools.get_episodes_impl(
          group_id=group_id,
          last_n=last_n,
      )

• rbt_mcp_server/graphiti_tools.py:160-196 (handler)

  Core handler implementation for get_episodes. Creates GraphitiClient, calls client.get_episodes, handles errors.

  async def get_episodes_impl(
      group_id: Optional[str] = None,
      last_n: int = 10,
  ) -> List[Dict[str, Any]]:
      """
      Get the most recent memory episodes for a specific group.
  
      @REQ: REQ-graphiti-chunk-mcp
      @BP: BP-graphiti-chunk-mcp
      @TASK: TASK-007-MCPTools
  
      Args:
          group_id: ID of the group to retrieve episodes from. If not provided, uses the default group_id.
          last_n: Number of most recent episodes to retrieve (default: 10)
  
      Returns:
          List of episode dictionaries
  
      Raises:
          ToolError: If retrieval operation fails
      """
      try:
          client = get_graphiti_client()
          async with client:
              group_ids = [group_id] if group_id else None
              results = await client.get_episodes(
                  last_n=last_n,
                  group_ids=group_ids,
              )
              return results
  
      except Exception as e:
          raise ToolError(
              "GET_EPISODES_ERROR",
              f"Failed to retrieve episodes: {str(e)}"
          ) from e

• rbt_mcp_server/chunking/graphiti_client.py:289-347 (helper)

  Low-level helper in GraphitiClient that retrieves episodes using graphiti.retrieve_episodes and formats them into dicts.

  async def get_episodes(
      self,
      reference_time: Optional[datetime] = None,
      last_n: int = 3,
      group_ids: Optional[List[str]] = None,
      source: Optional[EpisodeType] = None,
  ) -> List[Dict[str, Any]]:
      """
      Retrieve recent episodes from Graphiti.
  
      @REQ: REQ-graphiti-chunk-mcp
      @BP: BP-graphiti-chunk-mcp
      @TASK: TASK-005-GraphitiClient
  
      Args:
          reference_time: Reference timestamp (defaults to current time)
          last_n: Number of most recent episodes to retrieve (default: 3)
          group_ids: Optional list of group IDs to filter results
          source: Optional episode type filter
  
      Returns:
          List of episode dictionaries
  
      Raises:
          RuntimeError: If retrieval fails
      """
      if reference_time is None:
          reference_time = datetime.now()
  
      try:
          logger.debug(f"Retrieving last {last_n} episodes")
  
          episodes = await self.graphiti.retrieve_episodes(
              reference_time=reference_time,
              last_n=last_n,
              group_ids=group_ids,
              source=source,
          )
  
          # Convert EpisodicNode results to dictionaries
          episode_dicts = []
          for ep in episodes:
              episode_dicts.append({
                  "uuid": ep.uuid,
                  "name": ep.name,
                  "content": ep.content,
                  "source": ep.source.value if ep.source else None,
                  "source_description": ep.source_description,
                  "group_id": ep.group_id,
                  "created_at": ep.created_at,
                  "valid_at": ep.valid_at,
              })
  
          logger.info(f"Retrieved {len(episode_dicts)} episodes")
          return episode_dicts
  
      except Exception as e:
          logger.error(f"Failed to retrieve episodes: {e}")
          raise RuntimeError(f"Failed to retrieve episodes: {e}") from e