User Class

Overview

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

Properties

string

Unique identifier for the user.

dict

User metadata as a dictionary.

bool | None

Per-user merge conflict override. True = always raise merge conflicts for this user. False = never raise. None = inherit the project-level setting.

datetime

UTC timestamp when the user was created.

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
        merge_conflict_enabled=True  # optional: override merge conflict behaviour
    )
    print(f"Updated user ID: {user.user_id}")
except UserAlreadyExistsError as e:
    print(f"New user ID already exists: {e}")

dict

New metadata to replace the existing metadata. Completely replaces the old metadata.

string

Optional new user ID. Must be unique within your project.

bool

Per-user merge conflict override. True = always raise merge conflicts for this user. False = never raise. Pass None explicitly to reset to the project-level default.

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}")

integer

Automatically process the session after this many seconds of inactivity. Optional.

dict

Optional metadata to associate with the session.

datetime

Optional custom timestamp for when the session was created. Must be a timezone-aware datetime in UTC. Useful for benchmarking or importing historical data.

Returns: Session object Raises: UserNotFoundError, ValueError (if timestamp is not UTC)

The custom_created_at_utc parameter is particularly useful when:

Importing historical conversation data and preserving original timestamps
Running benchmarks with controlled temporal context
Migrating data from another system with existing timestamps

Example with Custom Timestamp

from datetime import datetime, timezone

user = client.get_user("user123")

# Create session with historical timestamp
historical_time = datetime(2025, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
session = user.create_session(
    custom_created_at_utc=historical_time,
    metadata={"imported": True, "source": "legacy_system"}
)

print(f"Session created with timestamp: {session.created_at}")

The timestamp must be timezone-aware and in UTC. Naive datetime objects or non-UTC timezones will raise a ValueError.

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}")

string

required

The UUID of the session to retrieve.

boolean

default:"True"

Whether to validate session existence via API before creating the instance. Set False only when user_id and session_id are already trusted.

Returns: Session object Raises: UserNotFoundError, SessionNotFoundError

Set validate=False to skip the SDK lookup request (GET /api/v1/users/{user_id}/sessions/{session_id}) when IDs are already trusted.

When validate=False, fields that require an API lookup (for example status, created_at, and metadata) are set to UNAVAILABLE until you call refresh(). Import UNAVAILABLE from recallrai.models when checking these values.

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}")

integer

Number of sessions to skip. Default: 0

integer

Maximum number of sessions to return. Default: 10

dict

Filter sessions by metadata fields.

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}")
        print(f"Event occurred: {mem.event_date_start} to {mem.event_date_end}")
        print(f"Recorded at: {mem.created_at}")
        
except InvalidCategoriesError as e:
    print(f"Invalid categories: {e.invalid_categories}")

list[string]

Filter by memory categories. Only memories matching these categories are returned.

list[string]

Filter by specific session IDs.

dict

Filter by session metadata.

integer

Number of memories to skip. Default: 0

integer

Maximum number of memories to return. Range: 1-200. Default: 20

boolean

Include version history for each memory. Default: True

boolean

Include related memories. Default: True

Returns: MemoryList object Raises: UserNotFoundError, InvalidCategoriesError

get_memory()

Retrieve a single memory by its ID.

from recallrai.exceptions import RecallrAIError

try:
    user = client.get_user("user123")
    memory = user.get_memory(
        memory_id="memory-uuid",
        include_previous_versions=True,
        include_connected_memories=True,
    )
    print(f"Content: {memory.content}")
    print(f"Categories: {memory.categories}")
    print(f"Version: {memory.version_number} of {memory.total_versions}")
except RecallrAIError as e:
    print(f"Error: {e}")

string

required

UUID of the memory to retrieve.

boolean

Include version history for the memory. Default: True

boolean

Include related memories. Default: True

Returns: UserMemoryItem object Raises: RecallrAIError

delete_memory()

Delete a specific memory version, with an option to also remove all previous versions in the chain.

from recallrai.exceptions import RecallrAIError

try:
    user = client.get_user("user123")

    # Delete only the specified memory version
    user.delete_memory(memory_id="memory-uuid")

    # Delete the specified version and all previous versions
    user.delete_memory(memory_id="memory-uuid", delete_previous_versions=True)
except RecallrAIError as e:
    print(f"Error: {e}")

string

required

UUID of the memory to delete. Can be any version in the version chain.

boolean

If True, deletes the specified version and all previous versions in the chain. If False (default), deletes only the specified version.

Deletion is permanent and cannot be undone. When delete_previous_versions=True, the entire version history up to and including the specified version is removed.

Returns: None Raises: RecallrAIError

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
event_date_start: UTC timestamp when the event started (actual event time, not when it was recorded)
event_date_end: UTC timestamp when the event ended (actual event time, not when it was recorded)
created_at: UTC timestamp when this memory version was created (when it was recorded in the system)
expired_at: UTC timestamp when this version expired — only set when viewing an expired/previous version
expiration_reason: Why this version was superseded (MERGE_CONFLICT, ADDITION_TO_EXISTING_MEMORY, TEMPORAL_CONFLICT) — only set for expired versions
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 an active (unresolved) conflict on this memory
merge_conflict_id: ID of the merge conflict that caused this memory to expire — only set when expiration_reason is MERGE_CONFLICT and a conflict record exists (manually resolved conflicts only)

Each MemoryVersionInfo object in previous_versions contains:

memory_id: ID of that specific version (can be passed to get_memory() for full details)
version_number: Version number (1 = oldest)
content: Content of that version
event_date_start / event_date_end: Event timestamps for that version
created_at: When that version was created
expired_at: When that version expired
expiration_reason: Why it was superseded
merge_conflict_id: Conflict that caused expiration, if applicable

The difference between event_date_start/event_date_end and created_at:

Event dates represent when the event actually occurred in the real world (e.g., “I met John on Monday”)
Created at represents when the memory was extracted and stored in the system

This distinction allows for better temporal reasoning when extracting memories from past conversations.

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}")

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}")

integer

Number of conflicts to skip. Default: 0

integer

Maximum number of conflicts to return. Default: 10

MergeConflictStatus

Filter by status: PENDING, IN_QUEUE, RESOLVING, RESOLVED, FAILED

string

Sort field: created_at or resolved_at. Default: created_at

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}")

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)

Working with Historical Data

When importing historical data or running benchmarks, you can preserve original timestamps:

from datetime import datetime, timezone
from recallrai import RecallrAI
from recallrai.models import MessageRole

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

# Create session with historical timestamp
historical_timestamp = datetime(2024, 12, 25, 14, 30, 0, tzinfo=timezone.utc)
session = user.create_session(
    custom_created_at_utc=historical_timestamp,
    metadata={"source": "import", "original_platform": "legacy_system"}
)

# Add messages from historical conversation
session.add_message(role=MessageRole.USER, content="What's the weather?")
session.add_message(role=MessageRole.ASSISTANT, content="It's sunny today!")

# Process the session - memories will use the historical timestamp for temporal context
session.process()

When sessions are processed with custom timestamps, the memory extraction and context retrieval use that timestamp instead of the current time. This ensures accurate temporal context for benchmarking and historical data analysis.

Python

Node.js

Overview

Properties

User Management Methods

update()

refresh()

delete()

Session Management Methods

create_session()

Example with Custom Timestamp

get_session()

list_sessions()

Memory Management Methods

list_memories()

get_memory()

delete_memory()

Memory Item Fields

Message Methods

get_last_n_messages()

Merge Conflict Methods

list_merge_conflicts()

get_merge_conflict()

Async User

Working with Historical Data

​Overview

​Properties

​User Management Methods

​update()

​refresh()

​delete()

​Session Management Methods

​create_session()

​Example with Custom Timestamp

​get_session()

​list_sessions()

​Memory Management Methods

​list_memories()

​get_memory()

​delete_memory()

​Memory Item Fields

​Message Methods

​get_last_n_messages()

​Merge Conflict Methods

​list_merge_conflicts()

​get_merge_conflict()

​Async User

​Working with Historical Data

Overview

Properties

User Management Methods

update()

refresh()

delete()

Session Management Methods

create_session()

Example with Custom Timestamp

get_session()

list_sessions()

Memory Management Methods

list_memories()

get_memory()

delete_memory()

Memory Item Fields

Message Methods

get_last_n_messages()

Merge Conflict Methods

list_merge_conflicts()

get_merge_conflict()

Async User

Working with Historical Data