Skip to main content

Overview

The User class represents a user in your RecallrAI project. It provides methods for managing sessions, memories, messages, and merge conflicts.

Properties

user_id
string
Unique identifier for the user.
metadata
dict
User metadata as a dictionary.
created_at
datetime
UTC timestamp when the user was created.
last_active_at
datetime
UTC timestamp of the user’s last activity.

User Management Methods

update()

Update the user’s metadata or ID.
from recallrai.exceptions import UserNotFoundError, UserAlreadyExistsError

try:
    user = client.get_user("user123")
    user.update(
        new_metadata={"name": "John Doe", "role": "admin"},
        new_user_id="john_doe"  # optional: change user ID
    )
    print(f"Updated user ID: {user.user_id}")
except UserAlreadyExistsError as e:
    print(f"New user ID already exists: {e}")
new_metadata
dict
New metadata to replace the existing metadata. Completely replaces the old metadata.
new_user_id
string
Optional new user ID. Must be unique within your project.
Returns: None (mutates the instance in place) Raises: UserNotFoundError, UserAlreadyExistsError

refresh()

Refresh the user instance with the latest data from the server.
user = client.get_user("john_doe")
user.refresh()
print(f"Latest metadata: {user.metadata}")
Returns: None (updates the instance in place) Raises: UserNotFoundError

delete()

Delete the user and all associated data.
from recallrai.exceptions import UserNotFoundError

try:
    user = client.get_user("john_doe")
    user.delete()
    print("User deleted successfully")
except UserNotFoundError as e:
    print(f"Error: {e}")
This permanently deletes the user and all associated sessions, memories, and messages. This action cannot be undone.
Returns: None Raises: UserNotFoundError

Session Management Methods

create_session()

Create a new session for the user.
from recallrai.exceptions import UserNotFoundError

try:
    user = client.get_user("user123")
    session = user.create_session(
        auto_process_after_seconds=600,
        metadata={"type": "chat", "channel": "web"}
    )
    print(f"Created session: {session.session_id}")
except UserNotFoundError as e:
    print(f"Error: {e}")
auto_process_after_seconds
integer
Automatically process the session after this many seconds of inactivity. Optional.
metadata
dict
Optional metadata to associate with the session.
Returns: Session object Raises: UserNotFoundError

get_session()

Retrieve an existing session by ID.
from recallrai.exceptions import SessionNotFoundError

try:
    user = client.get_user("user123")
    session = user.get_session(session_id="session-uuid")
    print(f"Session status: {session.status}")
except SessionNotFoundError as e:
    print(f"Error: {e}")
session_id
string
required
The UUID of the session to retrieve.
Returns: Session object Raises: UserNotFoundError, SessionNotFoundError

list_sessions()

List all sessions for the user with optional filtering.
from recallrai.models import SessionStatus

user = client.get_user("user123")
session_list = user.list_sessions(
    offset=0,
    limit=10,
    metadata_filter={"type": "chat"},
    status_filter=[SessionStatus.PENDING, SessionStatus.PROCESSING]
)

print(f"Total sessions: {session_list.total}")
for session in session_list.sessions:
    print(f"{session.session_id}: {session.status}")
offset
integer
Number of sessions to skip. Default: 0
limit
integer
Maximum number of sessions to return. Default: 10
metadata_filter
dict
Filter sessions by metadata fields.
status_filter
list[SessionStatus]
Filter by session status. Available statuses: PENDING, PROCESSING, PROCESSED, FAILED
Returns: SessionList object with sessions, total, and has_more fields

Memory Management Methods

list_memories()

List user memories with optional filtering.
from recallrai.exceptions import InvalidCategoriesError

try:
    user = client.get_user("user123")
    memories = user.list_memories(
        categories=["food_preferences", "allergies"],
        session_id_filter=["session-uuid-1", "session-uuid-2"],
        session_metadata_filter={"environment": "production"},
        offset=0,
        limit=20,
        include_previous_versions=True,
        include_connected_memories=True
    )
    
    for mem in memories.items:
        print(f"Memory: {mem.content}")
        print(f"Categories: {mem.categories}")
        print(f"Version: {mem.version_number} of {mem.total_versions}")
        
except InvalidCategoriesError as e:
    print(f"Invalid categories: {e.invalid_categories}")
categories
list[string]
Filter by memory categories. Only memories matching these categories are returned.
session_id_filter
list[string]
Filter by specific session IDs.
session_metadata_filter
dict
Filter by session metadata.
offset
integer
Number of memories to skip. Default: 0
limit
integer
Maximum number of memories to return. Range: 1-200. Default: 20
include_previous_versions
boolean
Include version history for each memory. Default: True
include_connected_memories
boolean
Include related memories. Default: True
Returns: MemoryList object Raises: UserNotFoundError, InvalidCategoriesError

Memory Item Fields

Each memory item contains:
  • memory_id: Unique identifier for the current version
  • categories: List of category strings
  • content: Current version’s content text
  • created_at: UTC timestamp when created
  • session_id: ID of the session that created this version
  • version_number: Current version number
  • total_versions: Total number of versions
  • has_previous_versions: Boolean indicating multiple versions exist
  • previous_versions: List of MemoryVersionInfo objects (optional)
  • connected_memories: List of MemoryRelationship objects (optional)
  • merge_conflict_in_progress: Boolean indicating active conflict

Message Methods

get_last_n_messages()

Retrieve the most recent messages for the user across all sessions.
user = client.get_user("user123")
messages = user.get_last_n_messages(n=5)

for msg in messages.messages:
    print(f"Session: {msg.session_id}")
    print(f"{msg.role.upper()}: {msg.content}")
n
integer
required
Number of recent messages to retrieve.
This is useful for chatbot applications where you need conversation context, such as WhatsApp bots where you want the last few messages to understand the ongoing conversation.
Returns: MessageList object with messages field Raises: UserNotFoundError

Merge Conflict Methods

list_merge_conflicts()

List merge conflicts for the user.
from recallrai.models import MergeConflictStatus

user = client.get_user("user123")
conflicts = user.list_merge_conflicts(
    offset=0,
    limit=10,
    status=MergeConflictStatus.PENDING,
    sort_by="created_at",
    sort_order="desc"
)

print(f"Total conflicts: {conflicts.total}")
for conflict in conflicts.conflicts:
    print(f"Conflict ID: {conflict.conflict_id}")
    print(f"Status: {conflict.status}")
offset
integer
Number of conflicts to skip. Default: 0
limit
integer
Maximum number of conflicts to return. Default: 10
status
MergeConflictStatus
Filter by status: PENDING, IN_QUEUE, RESOLVING, RESOLVED, FAILED
sort_by
string
Sort field: created_at or resolved_at. Default: created_at
sort_order
string
Sort order: asc or desc. Default: desc
Returns: MergeConflictList object

get_merge_conflict()

Get a specific merge conflict by ID.
from recallrai.exceptions import MergeConflictNotFoundError

try:
    user = client.get_user("user123")
    conflict = user.get_merge_conflict("conflict-uuid")
    print(f"Status: {conflict.status}")
    print(f"Clarifying questions: {len(conflict.clarifying_questions)}")
except MergeConflictNotFoundError as e:
    print(f"Error: {e}")
conflict_id
string
required
The UUID of the merge conflict to retrieve.
Returns: MergeConflict object Raises: UserNotFoundError, MergeConflictNotFoundError

Async User

For async applications, use AsyncUser:
from recallrai import AsyncRecallrAI

client = AsyncRecallrAI(api_key="rai_yourapikey", project_id="project-uuid")
user = await client.get_user("user123")

# All methods are the same, just use await
await user.update(new_metadata={"name": "Jane"})
await user.refresh()
session = await user.create_session()
memories = await user.list_memories(limit=10)