get_episodes
Retrieve recent memory episodes from a specific group to access historical data and track activity patterns for analysis.
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
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| group_id | No | ||
| last_n | No |
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
- 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