Repository files navigation

RecallrAI Python SDK

Official Python SDK for RecallrAI – a revolutionary contextual memory system that enables AI assistants to form meaningful connections between conversations, just like human memory.

Note: All datetime objects returned by the SDK are in UTC timezone.

Installation

Install the SDK via Poetry or pip:

poetry add recallrai
# or
pip install recallrai

Async Support

The SDK provides full async/await support for all operations! Use AsyncRecallrAI, AsyncUser, and AsyncSession for async applications. All usage patterns are identical to the sync versions, just with await keywords.

Initialization

Create a client instance with your API key and project ID:

fromrecallraiimportRecallrAIclient=RecallrAI(
api_key="rai_yourapikey",
project_id="project-uuid",
base_url="https://api.recallrai.com", # custom endpoint if applicabletimeout=60, # seconds
)

User Management

Create a User

fromrecallrai.exceptionsimportUserAlreadyExistsErrortry:
user=client.create_user(user_id="user123", metadata={"name": "John Doe"})
print(f"Created user: {user.user_id}")
print(f"User metadata: {user.metadata}")
print(f"Created at: {user.created_at}")
print(f"Merge conflict enabled: {user.merge_conflict_enabled}") # None = inherit project settingexceptUserAlreadyExistsErrorase:
print(f"Error: {e}")

Get a User

fromrecallrai.exceptionsimportUserNotFoundErrortry:
user=client.get_user("user123")
print(f"User metadata: {user.metadata}")
print(f"Last active: {user.last_active_at}")
exceptUserNotFoundErrorase:
print(f"Error: {e}")

If user_id is already trusted, you can skip the lookup request:

user=client.get_user("user123", validate=False)

List Users

user_list=client.list_users(offset=0, limit=10, metadata_filter={"role": "admin"})
print(f"Total users: {user_list.total}")
print(f"Has more users: {user_list.has_more}")
print("---")
foruinuser_list.users:
print(f"User ID: {u.user_id}")
print(f"Metadata: {u.metadata}")
print(f"Created at: {u.created_at}")
print(f"Last active: {u.last_active_at}")
print("---")

Update a User

fromrecallrai.exceptionsimportUserNotFoundError, UserAlreadyExistsErrortry:
user=client.get_user("user123")
# update() mutates the instance; no value is returneduser.update(
new_metadata={"name": "John Doe", "role": "admin"},
new_user_id="john_doe",
merge_conflict_enabled=True# override: always raise merge conflicts for this user
)
print(f"Updated user ID: {user.user_id}")
print(f"Updated metadata: {user.metadata}")
print(f"Merge conflict enabled: {user.merge_conflict_enabled}")
print(f"Last active: {user.last_active_at}")
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptUserAlreadyExistsErrorase:
print(f"Error: {e}")

Refresh User Instance

fromrecallrai.exceptionsimportUserNotFoundErrortry:
user=client.get_user("john_doe")
user.refresh()
print(f"Refreshed user metadata: {user.metadata}")
print(f"Last active: {user.last_active_at}")
exceptUserNotFoundErrorase:
print(f"Error: {e}")

Delete a User

fromrecallrai.exceptionsimportUserNotFoundErrortry:
user=client.get_user("john_doe")
user.delete()
print("User deleted successfully")
exceptUserNotFoundErrorase:
print(f"Error: {e}")

Session Management

Create a Session

fromrecallrai.exceptionsimportUserNotFoundErrorfromrecallrai.sessionimportSessiontry:
# First, get the useruser=client.get_user("user123")
# Create a session for the user.session: Session=user.create_session(
auto_process_after_seconds=600,
metadata={"type": "chat"}
)
print("Created session id:", session.session_id)
exceptUserNotFoundErrorase:
print(f"Error: {e}")

Get an Existing Session

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundErrortry:
# First, get the useruser=client.get_user("user123")
# Retrieve an existing session by its IDsession=user.get_session(session_id="session-uuid")
print("Session status:", session.status)
print("Session metadata:", session.metadata)
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")

If session_id is trusted, you can skip this lookup request:

session=user.get_session(session_id="session-uuid", validate=False)

Trusted IDs – Skip Validation Lookups

fromrecallrai.modelsimportRecallStrategy# Skips GET /api/v1/users/{user_id}user=client.get_user("user123", validate=False)
# Skips GET /api/v1/users/{user_id}/sessions/{session_id}session=user.get_session(session_id="session-uuid", validate=False)
# Goes directly to context retrievalcontext=session.get_context(recall_strategy=RecallStrategy.LOW_LATENCY)
print(context.context)

Use this only when IDs are already trusted by your system. This optimization skips SDK pre-validation calls.

When validate=False is used, unknown reference fields are set to UNAVAILABLE until you call refresh(). Import it from recallrai.models when you need to check for this sentinel.

Update a Session

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundErrortry:
# First, get the useruser=client.get_user("user123")
# Retrieve an existing session by its IDsession=user.get_session(session_id="session-uuid")
# Update session metadatasession.update(new_metadata={"type": "support_chat"})
print("Updated session metadata:", session.metadata)
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")

Refresh a Session

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundErrortry:
# First, get the useruser=client.get_user("user123")
# Retrieve an existing session by its IDsession=user.get_session(session_id="session-uuid")
# Refresh session data from the serversession.refresh()
print("Session status:", session.status)
print("Refreshed session metadata:", session.metadata)
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")

Delete a Session

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundErrortry:
user=client.get_user("user123")
session=user.get_session(session_id="session-uuid")
session.delete()
print("Session deleted successfully")
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")

List Sessions

fromrecallrai.modelsimportSessionStatusfromrecallrai.exceptionsimportUserNotFoundErrortry:
# First, get the useruser=client.get_user("user123")
# List sessions for this user with optional filterssession_list=user.list_sessions(
offset=0,
limit=10,
metadata_filter={"type": "chat"}, # optional: filter by session metadatastatus_filter=[SessionStatus.PENDING, SessionStatus.PROCESSING] # optional: filter by session status
)
print(f"Total sessions: {session_list.total}")
print(f"Has more sessions: {session_list.has_more}")
forsinsession_list.sessions:
print(s.session_id, s.status, s.metadata)
exceptUserNotFoundErrorase:
print(f"Error: {e}")

Session – Adding Messages

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundError, InvalidSessionStateErrorfromrecallrai.modelsimportMessageRoletry:
# Add a user messagesession.add_message(role=MessageRole.USER, content="Hello! How are you?")
# Add an assistant messagesession.add_message(role=MessageRole.ASSISTANT, content="I'm an assistant. How can I help you?")
# Available message roles:# - MessageRole.USER: Messages from the user/human# - MessageRole.ASSISTANT: Messages from the AI assistantexceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")
exceptInvalidSessionStateErrorase:
print(f"Error: {e}")

Session – Retrieving Context

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundErrorfromrecallrai.modelsimportRecallStrategytry:
# Get context with default parameterscontext=session.get_context()
print("Context:", context.context)
# Get context with specific recall strategycontext=session.get_context(recall_strategy=RecallStrategy.LOW_LATENCY)
print("Context:", context.context)
# Get context with custom memory retrieval parameterscontext=session.get_context(
recall_strategy=RecallStrategy.BALANCED,
min_top_k=10,
max_top_k=100,
memories_threshold=0.6,
summaries_threshold=0.5,
last_n_messages=20,
last_n_summaries=5,
timezone="America/Los_Angeles"# Optional: timezone for timestamp formatting, None for UTC
)
print("Context:", context.context)
# Get context with metadata detailscontext=session.get_context(include_metadata_ids=True)
ifcontext.metadata:
print("Memory IDs:", context.metadata.memory_ids)
print("Session IDs:", context.metadata.session_ids)
print("Vector Queries:", context.metadata.vector_search_queries)
print("Keywords:", context.metadata.keywords)
print("Summary Queries:", context.metadata.session_summaries_search_queries)
print("Date Filters:", context.metadata.date_range_filters)
print("Agent Reasoning:", context.metadata.agent_reasoning)
# Available recall strategies:# - RecallStrategy.LOW_LATENCY: Fast retrieval with basic relevance# - RecallStrategy.BALANCED: Good balance of speed and quality (default)# - RecallStrategy.AGENTIC: Agentic exploration for complex queries# Parameters:# - min_top_k: Minimum number of memories to return (default: 15, range: 5-50)# - max_top_k: Maximum number of memories to return (default: 50, range: 10-100)# - memories_threshold: Similarity threshold for memories (default: 0.6, range: 0.2-0.8)# - summaries_threshold: Similarity threshold for summaries (default: 0.5, range: 0.2-0.8)# - last_n_messages: Number of last messages to include in context (optional, range: 1-100)# - last_n_summaries: Number of last summaries to include in context (optional, range: 1-20)# - timezone: Timezone for formatting timestamps (optional, e.g., 'America/New_York', None for UTC)# - include_system_prompt: Whether to include the default system prompt of Recallr AI (default: True)exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")

Session – Streaming Context

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundErrorfromrecallrai.modelsimportRecallStrategytry:
foreventinsession.get_context_stream(
recall_strategy=RecallStrategy.BALANCED,
timezone="America/Los_Angeles",
):
ifevent.status_update_message:
print("Status:", event.status_update_message)
ifevent.metadata:
print("Metadata:", event.metadata)
ifevent.is_final:
ifevent.error_message:
print("Error:", event.error_message)
else:
print("Final context:", event.context)
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")

Session – Process Session

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundError, InvalidSessionStateErrortry:
session.process()
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")
exceptInvalidSessionStateErrorase:
print(f"Error: {e}")

Session – List Messages

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundErrortry:
# Paginated retrievalmessages=session.get_messages(offset=0, limit=50)
formsginmessages.messages:
print(f"{msg.role.value.upper()} (at {msg.timestamp}): {msg.content}")
print(f"Has more?: {messages.has_more}")
print(f"Total messages: {messages.total}")
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")

User Memories

List User Memories (with optional category filters)

fromrecallrai.exceptionsimportUserNotFoundError, InvalidCategoriesErrortry:
user=client.get_user("user123")
# List memories with all available optionsmemories=user.list_memories(
categories=["food_preferences", "allergies"], # optional: filter by categoriessession_id_filter=["session-uuid-1", "session-uuid-2"], # optional: filter by specific sessionssession_metadata_filter={"environment": "production"}, # optional: filter by session metadataoffset=0,
limit=20, # max 200include_previous_versions=True, # default: True - include version historyinclude_connected_memories=True, # default: True - include related memories
)
formeminmemories.items:
print(f"Memory ID: {mem.memory_id}")
print(f"Categories: {mem.categories}")
print(f"Content: {mem.content}")
print(f"Created at: {mem.created_at}")
print(f"Session ID: {mem.session_id}")
# Version informationprint(f"Version: {mem.version_number} of {mem.total_versions}")
print(f"Has previous versions: {mem.has_previous_versions}")
# Previous versions (if included)ifmem.previous_versions:
print(f"Previous versions: {len(mem.previous_versions)}")
forversioninmem.previous_versions:
print(f" - Version {version.version_number}: {version.content}")
print(f" Created: {version.created_at}, Expired: {version.expired_at}")
print(f" Expiration reason: {version.expiration_reason}")
# Connected memories (if included)ifmem.connected_memories:
print(f"Connected memories: {len(mem.connected_memories)}")
forconnectedinmem.connected_memories:
print(f" - {connected.memory_id}: {connected.content}")
# Merge conflict statusprint(f"Merge conflict in progress: {mem.merge_conflict_in_progress}")
print("---")
print(f"Has more?: {memories.has_more}")
print(f"Total memories: {memories.total}")
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptInvalidCategoriesErrorase:
print(f"Invalid categories: {e.invalid_categories}")
print(f"Error: {e}")

Memory Item Fields

Each memory item returned contains the following information:

  • memory_id: Unique identifier for the current/latest version of the memory
  • categories: List of category strings the memory belongs to
  • content: The current version's content text
  • created_at: Timestamp when the latest version was created
  • session_id: ID of the session that created this version
  • version_number: Which version this is (e.g., 3 means this is the 3rd version)
  • total_versions: Total number of versions that exist for this memory
  • has_previous_versions: Boolean indicating if total_versions > 1
  • previous_versions (optional): List of MemoryVersionInfo objects containing:
    • version_number: Sequential version number (1 = oldest)
    • content: Content of that version
    • created_at: When this version was created
    • expired_at: When this version expired
    • expiration_reason: Why it expired (e.g., new version created)
  • connected_memories (optional): List of MemoryRelationship objects containing:
    • memory_id: ID of the connected memory
    • content: Brief content for context
  • merge_conflict_in_progress: Boolean indicating if this memory has an active merge conflict

User Messages

Get Last N Messages

Retrieve the most recent messages for a user across all their sessions. This is particularly useful for chatbot applications where you need conversation context, such as WhatsApp support bots where you want to pass the last few messages to understand the ongoing conversation.

fromrecallrai.exceptionsimportUserNotFoundErrortry:
user=client.get_user("user123")
# Fetch last N messages (e.g., last 5)messages=user.get_last_n_messages(n=5)
formsginmessages.messages:
print(f"Session ID: {msg.session_id}")
print(f"{msg.role.upper()} (at {msg.timestamp}): {msg.content}")
print("---")
exceptUserNotFoundErrorase:
print(f"Error: {e}")

Merge Conflict Management

When RecallrAI processes sessions, it may detect conflicts between new memories and existing ones. The SDK provides comprehensive tools to handle these merge conflicts, allowing you to guide the resolution process through clarifying questions.

List Merge Conflicts

fromrecallrai.exceptionsimportUserNotFoundErrorfromrecallrai.modelsimportMergeConflictStatustry:
user=client.get_user("user123")
# List all merge conflicts (with optional status filter)conflicts=user.list_merge_conflicts(
offset=0,
limit=10,
status=MergeConflictStatus.PENDING, # optional: filter by statussort_by="created_at", # created_at, resolved_atsort_order="desc", # asc, desc
)
print(f"Total conflicts: {conflicts.total}")
print(f"Has more: {conflicts.has_more}")
forconfinconflicts.conflicts:
print(f"Conflict ID: {conf.conflict_id}")
print(f"Status: {conf.status}")
print(f"New memory: {conf.proposed_memory_content}")
print(f"Conflicting memories: {len(conf.conflicting_memories)}")
print(f"Questions: {len(conf.clarifying_questions)}")
print(f"Created at: {conf.created_at}")
print("---")
exceptUserNotFoundErrorase:
print(f"Error: {e}")

Get a Specific Merge Conflict

fromrecallrai.exceptionsimportUserNotFoundError, MergeConflictNotFoundErrortry:
user=client.get_user("user123")
conflict=user.get_merge_conflict("conflict-uuid")
print(f"Conflict ID: {conflict.conflict_id}")
print(f"Status: {conflict.status.value}")
print(f"New memory content: {conflict.proposed_memory_content}")
# Examine conflicting memoriesprint("\nConflicting memories:")
formeminconflict.conflicting_memories:
print(f" Content: {mem.content}")
print(f" Reason: {mem.reason}")
print()
# View clarifying questionsprint("Clarifying questions:")
forquesinconflict.clarifying_questions:
print(f" Question: {ques.question}")
print(f" Options: {ques.options}")
print()
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptMergeConflictNotFoundErrorase:
print(f"Error: {e}")

Resolve a Merge Conflict

fromrecallrai.exceptionsimport (
UserNotFoundError, MergeConflictNotFoundError,
MergeConflictAlreadyResolvedError,
MergeConflictInvalidQuestionsError,
MergeConflictMissingAnswersError,
MergeConflictInvalidAnswerError,
ValidationError
)
fromrecallrai.modelsimportMergeConflictAnswertry:
user=client.get_user("user123")
conflict=user.get_merge_conflict("conflict-uuid")
# Prepare answers to the clarifying questionsanswers= []
forquesinconflict.clarifying_questions:
print(f" Question: {ques.question}")
print(f" Options: {ques.options}")
print()
answer=MergeConflictAnswer(
question=ques.question,
answer=ques.options[0], # Select first optionmessage="User prefers this option based on recent conversation"
)
answers.append(answer)
# Resolve the conflictconflict.resolve(answers)
print(f"Conflict resolved! Status: {conflict.status}")
print(f"Resolved at: {conflict.resolved_at}")
ifconflict.resolution_data:
print(f"Resolution data: {conflict.resolution_data}")
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptMergeConflictNotFoundErrorase:
print(f"Error: {e}")
exceptMergeConflictAlreadyResolvedErrorase:
print(f"Error: {e}")
exceptMergeConflictInvalidQuestionsErrorase:
print(f"Error: {e}")
ife.invalid_questions:
print(f"Invalid questions: {e.invalid_questions}")
exceptMergeConflictMissingAnswersErrorase:
print(f"Error: {e}")
ife.missing_questions:
print(f"Missing answers for: {e.missing_questions}")
exceptMergeConflictInvalidAnswerErrorase:
print(f"Error: {e}")
ife.questionande.valid_options:
print(f"Question: {e.question}")
print(f"Valid options: {e.valid_options}")
exceptValidationErrorase:
print(f"Error: {e}")

Refresh Merge Conflict Data

fromrecallrai.exceptionsimportUserNotFoundError, MergeConflictNotFoundErrortry:
user=client.get_user("user123")
conflict=user.get_merge_conflict("conflict-uuid")
# Refresh to get latest status from serverconflict.refresh()
print(f"Current status: {conflict.status}")
print(f"Last updated: {conflict.resolved_at}")
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptMergeConflictNotFoundErrorase:
print(f"Error: {e}")

Working with Merge Conflict Statuses

The merge conflict system uses several status values to track the lifecycle of conflicts:

  • PENDING: Conflict detected and waiting for resolution
  • IN_QUEUE: Conflict is queued for automated processing
  • RESOLVING: Conflict is being processed
  • RESOLVED: Conflict has been successfully resolved
  • FAILED: Conflict resolution failed

Example Usage with LLMs

importopenaifromrecallraiimportRecallrAIfromrecallrai.exceptionsimportUserNotFoundErrorfromrecallrai.modelsimportMessageRole# Initialize RecallrAI and OpenAI clientsrai_client=RecallrAI(
api_key="rai_yourapikey", project_id="your-project-uuid"
)
oai_client=openai.OpenAI(api_key="your-openai-api-key")
defchat_with_memory(user_id, session_id=None):
# Get or create usertry:
user=rai_client.get_user(user_id)
exceptUserNotFoundError:
user=rai_client.create_user(user_id)
# Create a new session or get an existing oneifsession_id:
session=user.get_session(session_id=session_id)
else:
session=user.create_session(auto_process_after_seconds=1800)
print(f"Created new session: {session.session_id}")
print("Chat session started. Type 'exit' to end the conversation.")
whileTrue:
# Get user inputuser_message=input("You: ")
ifuser_message.lower() =='exit':
break# Add the user message to RecallrAIsession.add_message(role=MessageRole.USER, content=user_message)
# Get context from RecallrAI after adding the user message# You can specify a recall strategy for different performance/quality trade-offs# Additional parameters like min_top_k, max_top_k, memories_threshold, summaries_threshold, last_n_messages, last_n_summaries are availablecontext=session.get_context() # Uses default BALANCED strategy# Create a system prompt that includes the contextsystem_prompt="You are a helpful assistant"+context.context# Get previous messagesmessages=session.get_messages(offset=0, limit=50)
previous_messages= [{"role": message.role, "content": message.content} formessageinmessages.messages]
# Call the LLM with the system prompt and conversation historyresponse=oai_client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": system_prompt},
**previous_messages,
],
temperature=0.7
)
assistant_message=response.choices[0].message.content# Print the assistant's responseprint(f"Assistant: {assistant_message}")
# Add the assistant's response to RecallrAIsession.add_message(role=MessageRole.ASSISTANT, content=assistant_message)
# Process the session at the end of the conversationprint("Processing session to update memory...")
session.process()
print(f"Session ended. Session ID: {session.session_id}")
returnsession.session_id# Example usageif__name__=="__main__":
user_id="user123"# To continue a previous session, uncomment below and provide the session ID# previous_session_id = "previously-saved-session-uuid"# session_id = chat_with_memory(user_id, previous_session_id)# Start a new sessionsession_id=chat_with_memory(user_id)
print(f"To continue this conversation later, use session ID: {session_id}")

Exception Handling

The RecallrAI SDK implements a comprehensive exception hierarchy to help you handle different error scenarios gracefully:

Base Exception

  • RecallrAIError: The base exception for all SDK-specific errors. All other exceptions inherit from this.

Authentication Errors

  • AuthenticationError: Raised when there's an issue with your API key or project ID authentication.

Network-Related Errors

  • TimeoutError: Occurs when a request takes too long to complete.
  • ConnectionError: Happens when the SDK cannot establish a connection to the RecallrAI API.

Server Errors

  • InternalServerError: Raised when the RecallrAI API returns a 5xx error code.
  • RateLimitError: Raised when the API rate limit has been exceeded (HTTP 429). When available, the retry_after value is provided in the exception details.

User-Related Errors

  • UserNotFoundError: Raised when attempting to access a user that doesn't exist.
  • UserAlreadyExistsError: Occurs when creating a user with an ID that already exists.
  • InvalidCategoriesError: Raised when filtering user memories by categories that do not exist in the project. The exception contains the list of invalid categories in e.invalid_categories.

Session-Related Errors

  • SessionNotFoundError: Raised when attempting to access a non-existent session.
  • InvalidSessionStateError: Occurs when performing an operation that's not valid for the current session state (e.g., adding a message to a processed session).

Merge Conflict-Related Errors

  • MergeConflictError: Base class for merge conflict-related exceptions.
  • MergeConflictNotFoundError: Raised when attempting to access a merge conflict that doesn't exist.
  • MergeConflictAlreadyResolvedError: Occurs when trying to resolve a merge conflict that has already been processed.
  • MergeConflictInvalidQuestionsError: Raised when the provided questions don't match the original clarifying questions.
  • MergeConflictMissingAnswersError: Occurs when not all required clarifying questions have been answered.
  • MergeConflictInvalidAnswerError: Raised when an answer is not one of the valid options for a question.

Input Validation Errors

  • ValidationError: Raised when provided data doesn't meet the required format or constraints.

Importing Exceptions

You can import exceptions directly from the recallrai.exceptions module:

# Import specific exceptionsfromrecallrai.exceptionsimport (
UserNotFoundError, SessionNotFoundError,
InvalidCategoriesError,
MergeConflictNotFoundError,
MergeConflictAlreadyResolvedError,
MergeConflictInvalidQuestionsError,
MergeConflictMissingAnswersError,
MergeConflictInvalidAnswerError,
)
# Import all exceptionsfromrecallrai.exceptionsimport (
RecallrAIError,
AuthenticationError,
TimeoutError,
ConnectionError,
InternalServerError,
RateLimitError,
SessionNotFoundError, InvalidSessionStateError,
UserNotFoundError, UserAlreadyExistsError,
InvalidCategoriesError,
ValidationError,
MergeConflictError,
MergeConflictNotFoundError,
MergeConflictAlreadyResolvedError,
MergeConflictInvalidQuestionsError,
MergeConflictMissingAnswersError,
MergeConflictInvalidAnswerError,
)

Exception Hierarchy Diagram

flowchart LR
%% Title
title[Errors and Exceptions Hierarchy]
style title fill:none,stroke:none
%% Base exception class
Exception[Exception]
RecallrAIError[RecallrAIError]
%% First level exceptions
AuthenticationError[AuthenticationError]
NetworkError[NetworkError]
ServerError[ServerError]
UserError[UserError]
SessionError[SessionError]
MergeConflictError[MergeConflictError]
ValidationError[ValidationError]
%% Second level exceptions - NetworkError children
TimeoutError[TimeoutError]
ConnectionError[ConnectionError]
%% Second level exceptions - ServerError children
InternalServerError[Internal ServerError]
RateLimitError[RateLimitError]
%% Second level exceptions - UserError children
UserNotFoundError[User NotFound Error]
UserAlreadyExistsError[User AlreadyExists Error]
%% Second level exceptions - SessionError children
InvalidSessionStateError[Invalid SessionState Error]
SessionNotFoundError[Session NotFound Error]
%% Second level exceptions - MergeConflictError children
MergeConflictNotFoundError[MergeConflict NotFound Error]
MergeConflictAlreadyResolvedError[MergeConflict AlreadyResolved Error]
MergeConflictInvalidQuestionsError[MergeConflict InvalidQuestions Error]
MergeConflictMissingAnswersError[MergeConflict MissingAnswers Error]
MergeConflictInvalidAnswerError[MergeConflict InvalidAnswer Error]
%% Connect parent to base
Exception --> RecallrAIError
%% Connect base to first level
RecallrAIError --> AuthenticationError
RecallrAIError --> NetworkError
RecallrAIError --> ServerError
RecallrAIError --> UserError
RecallrAIError --> SessionError
RecallrAIError --> MergeConflictError
RecallrAIError --> ValidationError
%% Connect first level to second level
NetworkError --> TimeoutError
NetworkError --> ConnectionError
ServerError --> InternalServerError
ServerError --> RateLimitError
UserError --> UserNotFoundError
UserError --> UserAlreadyExistsError
SessionError --> InvalidSessionStateError
SessionError --> SessionNotFoundError
MergeConflictError --> MergeConflictNotFoundError
MergeConflictError --> MergeConflictAlreadyResolvedError
MergeConflictError --> MergeConflictInvalidQuestionsError
MergeConflictError --> MergeConflictMissingAnswersError
MergeConflictError --> MergeConflictInvalidAnswerError
Loading

Best Practices for Error Handling

When implementing error handling with the RecallrAI SDK, consider these best practices:

  1. Handle specific exceptions first: Catch more specific exceptions before general ones.

    try:
    # SDK operationexceptUserNotFoundError:
    # Specific handlingexceptRecallrAIError:
    # General fallback
  2. Implement retry logic for transient errors: Network and timeout errors might be temporary.

  3. Log detailed error information: Exceptions contain useful information for debugging.

  4. Handle common user flows: For example, check if a user exists before operations, or create them if they don't:

    try:
    user=client.get_user(user_id)
    exceptUserNotFoundError:
    user=client.create_user(user_id)

For more detailed information on specific exceptions, refer to the API documentation.

Conclusion

This README outlines the basic usage of the RecallrAI SDK functions for user and session management. For additional documentation and advanced usage, please see the official documentation or the source code repository on GitHub.

About

Official Python SDK for RecallrAI – a revolutionary contextual memory system that enables AI assistants to form meaningful connections between conversations, just like human memory.

Topics

Resources

Stars

3 stars

Watchers

0 watching

Forks

Releases

Packages

Used by

Contributors

Languages

, 'i'); if (__m === '*' || __re.test(location.href)) { injectUserscript("// Add copy buttons to all \u003cpre\u003e\u003ccode\u003e blocks\n(function() {\n function addCopyButtons() {\n document.querySelectorAll('pre code').forEach(function(codeBlock) {\n if (codeBlock.parentElement.hasAttribute('data-copy-added')) return;\n codeBlock.parentElement.setAttribute('data-copy-added', 'true');\n \n var btn = document.createElement('button');\n btn.textContent = 'Copy';\n btn.style.cssText = 'position:absolute;top:4px;right:4px;padding:2px 8px;font-size:11px;background:#4ecdc4;border:none;border-radius:4px;color:#1a1a2e;cursor:pointer;opacity:0.7;transition:opacity 0.2s;';\n btn.onmouseover = function() { this.style.opacity = '1'; };\n btn.onmouseout = function() { this.style.opacity = '0.7'; };\n btn.onclick = function() {\n navigator.clipboard.writeText(codeBlock.textContent).then(function() {\n btn.textContent = 'Copied!';\n setTimeout(function() { btn.textContent = 'Copy'; }, 1500);\n });\n };\n codeBlock.parentElement.style.position = 'relative';\n codeBlock.parentElement.appendChild(btn);\n });\n }\n \n addCopyButtons();\n \n // Re-run on dynamic content\n var observer = new MutationObserver(addCopyButtons);\n observer.observe(document.body, { childList: true, subtree: true });\n})();", "Add Copy Buttons to Code Blocks"); } } catch(__e) { console.warn('[Userscript:Add Copy Buttons to Code Blocks]', __e); } })(); (function(){ try { var __m = "github.com"; var __re = new RegExp('^' + "github\\.com" + '
Skip to content

Repository files navigation

RecallrAI Python SDK

Official Python SDK for RecallrAI – a revolutionary contextual memory system that enables AI assistants to form meaningful connections between conversations, just like human memory.

Note: All datetime objects returned by the SDK are in UTC timezone.

Installation

Install the SDK via Poetry or pip:

poetry add recallrai
# or
pip install recallrai

Async Support

The SDK provides full async/await support for all operations! Use AsyncRecallrAI, AsyncUser, and AsyncSession for async applications. All usage patterns are identical to the sync versions, just with await keywords.

Initialization

Create a client instance with your API key and project ID:

fromrecallraiimportRecallrAIclient=RecallrAI(
api_key="rai_yourapikey",
project_id="project-uuid",
base_url="https://api.recallrai.com", # custom endpoint if applicabletimeout=60, # seconds
)

User Management

Create a User

fromrecallrai.exceptionsimportUserAlreadyExistsErrortry:
user=client.create_user(user_id="user123", metadata={"name": "John Doe"})
print(f"Created user: {user.user_id}")
print(f"User metadata: {user.metadata}")
print(f"Created at: {user.created_at}")
print(f"Merge conflict enabled: {user.merge_conflict_enabled}") # None = inherit project settingexceptUserAlreadyExistsErrorase:
print(f"Error: {e}")

Get a User

fromrecallrai.exceptionsimportUserNotFoundErrortry:
user=client.get_user("user123")
print(f"User metadata: {user.metadata}")
print(f"Last active: {user.last_active_at}")
exceptUserNotFoundErrorase:
print(f"Error: {e}")

If user_id is already trusted, you can skip the lookup request:

user=client.get_user("user123", validate=False)

List Users

user_list=client.list_users(offset=0, limit=10, metadata_filter={"role": "admin"})
print(f"Total users: {user_list.total}")
print(f"Has more users: {user_list.has_more}")
print("---")
foruinuser_list.users:
print(f"User ID: {u.user_id}")
print(f"Metadata: {u.metadata}")
print(f"Created at: {u.created_at}")
print(f"Last active: {u.last_active_at}")
print("---")

Update a User

fromrecallrai.exceptionsimportUserNotFoundError, UserAlreadyExistsErrortry:
user=client.get_user("user123")
# update() mutates the instance; no value is returneduser.update(
new_metadata={"name": "John Doe", "role": "admin"},
new_user_id="john_doe",
merge_conflict_enabled=True# override: always raise merge conflicts for this user
)
print(f"Updated user ID: {user.user_id}")
print(f"Updated metadata: {user.metadata}")
print(f"Merge conflict enabled: {user.merge_conflict_enabled}")
print(f"Last active: {user.last_active_at}")
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptUserAlreadyExistsErrorase:
print(f"Error: {e}")

Refresh User Instance

fromrecallrai.exceptionsimportUserNotFoundErrortry:
user=client.get_user("john_doe")
user.refresh()
print(f"Refreshed user metadata: {user.metadata}")
print(f"Last active: {user.last_active_at}")
exceptUserNotFoundErrorase:
print(f"Error: {e}")

Delete a User

fromrecallrai.exceptionsimportUserNotFoundErrortry:
user=client.get_user("john_doe")
user.delete()
print("User deleted successfully")
exceptUserNotFoundErrorase:
print(f"Error: {e}")

Session Management

Create a Session

fromrecallrai.exceptionsimportUserNotFoundErrorfromrecallrai.sessionimportSessiontry:
# First, get the useruser=client.get_user("user123")
# Create a session for the user.session: Session=user.create_session(
auto_process_after_seconds=600,
metadata={"type": "chat"}
)
print("Created session id:", session.session_id)
exceptUserNotFoundErrorase:
print(f"Error: {e}")

Get an Existing Session

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundErrortry:
# First, get the useruser=client.get_user("user123")
# Retrieve an existing session by its IDsession=user.get_session(session_id="session-uuid")
print("Session status:", session.status)
print("Session metadata:", session.metadata)
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")

If session_id is trusted, you can skip this lookup request:

session=user.get_session(session_id="session-uuid", validate=False)

Trusted IDs – Skip Validation Lookups

fromrecallrai.modelsimportRecallStrategy# Skips GET /api/v1/users/{user_id}user=client.get_user("user123", validate=False)
# Skips GET /api/v1/users/{user_id}/sessions/{session_id}session=user.get_session(session_id="session-uuid", validate=False)
# Goes directly to context retrievalcontext=session.get_context(recall_strategy=RecallStrategy.LOW_LATENCY)
print(context.context)

Use this only when IDs are already trusted by your system. This optimization skips SDK pre-validation calls.

When validate=False is used, unknown reference fields are set to UNAVAILABLE until you call refresh(). Import it from recallrai.models when you need to check for this sentinel.

Update a Session

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundErrortry:
# First, get the useruser=client.get_user("user123")
# Retrieve an existing session by its IDsession=user.get_session(session_id="session-uuid")
# Update session metadatasession.update(new_metadata={"type": "support_chat"})
print("Updated session metadata:", session.metadata)
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")

Refresh a Session

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundErrortry:
# First, get the useruser=client.get_user("user123")
# Retrieve an existing session by its IDsession=user.get_session(session_id="session-uuid")
# Refresh session data from the serversession.refresh()
print("Session status:", session.status)
print("Refreshed session metadata:", session.metadata)
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")

Delete a Session

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundErrortry:
user=client.get_user("user123")
session=user.get_session(session_id="session-uuid")
session.delete()
print("Session deleted successfully")
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")

List Sessions

fromrecallrai.modelsimportSessionStatusfromrecallrai.exceptionsimportUserNotFoundErrortry:
# First, get the useruser=client.get_user("user123")
# List sessions for this user with optional filterssession_list=user.list_sessions(
offset=0,
limit=10,
metadata_filter={"type": "chat"}, # optional: filter by session metadatastatus_filter=[SessionStatus.PENDING, SessionStatus.PROCESSING] # optional: filter by session status
)
print(f"Total sessions: {session_list.total}")
print(f"Has more sessions: {session_list.has_more}")
forsinsession_list.sessions:
print(s.session_id, s.status, s.metadata)
exceptUserNotFoundErrorase:
print(f"Error: {e}")

Session – Adding Messages

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundError, InvalidSessionStateErrorfromrecallrai.modelsimportMessageRoletry:
# Add a user messagesession.add_message(role=MessageRole.USER, content="Hello! How are you?")
# Add an assistant messagesession.add_message(role=MessageRole.ASSISTANT, content="I'm an assistant. How can I help you?")
# Available message roles:# - MessageRole.USER: Messages from the user/human# - MessageRole.ASSISTANT: Messages from the AI assistantexceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")
exceptInvalidSessionStateErrorase:
print(f"Error: {e}")

Session – Retrieving Context

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundErrorfromrecallrai.modelsimportRecallStrategytry:
# Get context with default parameterscontext=session.get_context()
print("Context:", context.context)
# Get context with specific recall strategycontext=session.get_context(recall_strategy=RecallStrategy.LOW_LATENCY)
print("Context:", context.context)
# Get context with custom memory retrieval parameterscontext=session.get_context(
recall_strategy=RecallStrategy.BALANCED,
min_top_k=10,
max_top_k=100,
memories_threshold=0.6,
summaries_threshold=0.5,
last_n_messages=20,
last_n_summaries=5,
timezone="America/Los_Angeles"# Optional: timezone for timestamp formatting, None for UTC
)
print("Context:", context.context)
# Get context with metadata detailscontext=session.get_context(include_metadata_ids=True)
ifcontext.metadata:
print("Memory IDs:", context.metadata.memory_ids)
print("Session IDs:", context.metadata.session_ids)
print("Vector Queries:", context.metadata.vector_search_queries)
print("Keywords:", context.metadata.keywords)
print("Summary Queries:", context.metadata.session_summaries_search_queries)
print("Date Filters:", context.metadata.date_range_filters)
print("Agent Reasoning:", context.metadata.agent_reasoning)
# Available recall strategies:# - RecallStrategy.LOW_LATENCY: Fast retrieval with basic relevance# - RecallStrategy.BALANCED: Good balance of speed and quality (default)# - RecallStrategy.AGENTIC: Agentic exploration for complex queries# Parameters:# - min_top_k: Minimum number of memories to return (default: 15, range: 5-50)# - max_top_k: Maximum number of memories to return (default: 50, range: 10-100)# - memories_threshold: Similarity threshold for memories (default: 0.6, range: 0.2-0.8)# - summaries_threshold: Similarity threshold for summaries (default: 0.5, range: 0.2-0.8)# - last_n_messages: Number of last messages to include in context (optional, range: 1-100)# - last_n_summaries: Number of last summaries to include in context (optional, range: 1-20)# - timezone: Timezone for formatting timestamps (optional, e.g., 'America/New_York', None for UTC)# - include_system_prompt: Whether to include the default system prompt of Recallr AI (default: True)exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")

Session – Streaming Context

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundErrorfromrecallrai.modelsimportRecallStrategytry:
foreventinsession.get_context_stream(
recall_strategy=RecallStrategy.BALANCED,
timezone="America/Los_Angeles",
):
ifevent.status_update_message:
print("Status:", event.status_update_message)
ifevent.metadata:
print("Metadata:", event.metadata)
ifevent.is_final:
ifevent.error_message:
print("Error:", event.error_message)
else:
print("Final context:", event.context)
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")

Session – Process Session

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundError, InvalidSessionStateErrortry:
session.process()
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")
exceptInvalidSessionStateErrorase:
print(f"Error: {e}")

Session – List Messages

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundErrortry:
# Paginated retrievalmessages=session.get_messages(offset=0, limit=50)
formsginmessages.messages:
print(f"{msg.role.value.upper()} (at {msg.timestamp}): {msg.content}")
print(f"Has more?: {messages.has_more}")
print(f"Total messages: {messages.total}")
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")

User Memories

List User Memories (with optional category filters)

fromrecallrai.exceptionsimportUserNotFoundError, InvalidCategoriesErrortry:
user=client.get_user("user123")
# List memories with all available optionsmemories=user.list_memories(
categories=["food_preferences", "allergies"], # optional: filter by categoriessession_id_filter=["session-uuid-1", "session-uuid-2"], # optional: filter by specific sessionssession_metadata_filter={"environment": "production"}, # optional: filter by session metadataoffset=0,
limit=20, # max 200include_previous_versions=True, # default: True - include version historyinclude_connected_memories=True, # default: True - include related memories
)
formeminmemories.items:
print(f"Memory ID: {mem.memory_id}")
print(f"Categories: {mem.categories}")
print(f"Content: {mem.content}")
print(f"Created at: {mem.created_at}")
print(f"Session ID: {mem.session_id}")
# Version informationprint(f"Version: {mem.version_number} of {mem.total_versions}")
print(f"Has previous versions: {mem.has_previous_versions}")
# Previous versions (if included)ifmem.previous_versions:
print(f"Previous versions: {len(mem.previous_versions)}")
forversioninmem.previous_versions:
print(f" - Version {version.version_number}: {version.content}")
print(f" Created: {version.created_at}, Expired: {version.expired_at}")
print(f" Expiration reason: {version.expiration_reason}")
# Connected memories (if included)ifmem.connected_memories:
print(f"Connected memories: {len(mem.connected_memories)}")
forconnectedinmem.connected_memories:
print(f" - {connected.memory_id}: {connected.content}")
# Merge conflict statusprint(f"Merge conflict in progress: {mem.merge_conflict_in_progress}")
print("---")
print(f"Has more?: {memories.has_more}")
print(f"Total memories: {memories.total}")
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptInvalidCategoriesErrorase:
print(f"Invalid categories: {e.invalid_categories}")
print(f"Error: {e}")

Memory Item Fields

Each memory item returned contains the following information:

  • memory_id: Unique identifier for the current/latest version of the memory
  • categories: List of category strings the memory belongs to
  • content: The current version's content text
  • created_at: Timestamp when the latest version was created
  • session_id: ID of the session that created this version
  • version_number: Which version this is (e.g., 3 means this is the 3rd version)
  • total_versions: Total number of versions that exist for this memory
  • has_previous_versions: Boolean indicating if total_versions > 1
  • previous_versions (optional): List of MemoryVersionInfo objects containing:
    • version_number: Sequential version number (1 = oldest)
    • content: Content of that version
    • created_at: When this version was created
    • expired_at: When this version expired
    • expiration_reason: Why it expired (e.g., new version created)
  • connected_memories (optional): List of MemoryRelationship objects containing:
    • memory_id: ID of the connected memory
    • content: Brief content for context
  • merge_conflict_in_progress: Boolean indicating if this memory has an active merge conflict

User Messages

Get Last N Messages

Retrieve the most recent messages for a user across all their sessions. This is particularly useful for chatbot applications where you need conversation context, such as WhatsApp support bots where you want to pass the last few messages to understand the ongoing conversation.

fromrecallrai.exceptionsimportUserNotFoundErrortry:
user=client.get_user("user123")
# Fetch last N messages (e.g., last 5)messages=user.get_last_n_messages(n=5)
formsginmessages.messages:
print(f"Session ID: {msg.session_id}")
print(f"{msg.role.upper()} (at {msg.timestamp}): {msg.content}")
print("---")
exceptUserNotFoundErrorase:
print(f"Error: {e}")

Merge Conflict Management

When RecallrAI processes sessions, it may detect conflicts between new memories and existing ones. The SDK provides comprehensive tools to handle these merge conflicts, allowing you to guide the resolution process through clarifying questions.

List Merge Conflicts

fromrecallrai.exceptionsimportUserNotFoundErrorfromrecallrai.modelsimportMergeConflictStatustry:
user=client.get_user("user123")
# List all merge conflicts (with optional status filter)conflicts=user.list_merge_conflicts(
offset=0,
limit=10,
status=MergeConflictStatus.PENDING, # optional: filter by statussort_by="created_at", # created_at, resolved_atsort_order="desc", # asc, desc
)
print(f"Total conflicts: {conflicts.total}")
print(f"Has more: {conflicts.has_more}")
forconfinconflicts.conflicts:
print(f"Conflict ID: {conf.conflict_id}")
print(f"Status: {conf.status}")
print(f"New memory: {conf.proposed_memory_content}")
print(f"Conflicting memories: {len(conf.conflicting_memories)}")
print(f"Questions: {len(conf.clarifying_questions)}")
print(f"Created at: {conf.created_at}")
print("---")
exceptUserNotFoundErrorase:
print(f"Error: {e}")

Get a Specific Merge Conflict

fromrecallrai.exceptionsimportUserNotFoundError, MergeConflictNotFoundErrortry:
user=client.get_user("user123")
conflict=user.get_merge_conflict("conflict-uuid")
print(f"Conflict ID: {conflict.conflict_id}")
print(f"Status: {conflict.status.value}")
print(f"New memory content: {conflict.proposed_memory_content}")
# Examine conflicting memoriesprint("\nConflicting memories:")
formeminconflict.conflicting_memories:
print(f" Content: {mem.content}")
print(f" Reason: {mem.reason}")
print()
# View clarifying questionsprint("Clarifying questions:")
forquesinconflict.clarifying_questions:
print(f" Question: {ques.question}")
print(f" Options: {ques.options}")
print()
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptMergeConflictNotFoundErrorase:
print(f"Error: {e}")

Resolve a Merge Conflict

fromrecallrai.exceptionsimport (
UserNotFoundError, MergeConflictNotFoundError,
MergeConflictAlreadyResolvedError,
MergeConflictInvalidQuestionsError,
MergeConflictMissingAnswersError,
MergeConflictInvalidAnswerError,
ValidationError
)
fromrecallrai.modelsimportMergeConflictAnswertry:
user=client.get_user("user123")
conflict=user.get_merge_conflict("conflict-uuid")
# Prepare answers to the clarifying questionsanswers= []
forquesinconflict.clarifying_questions:
print(f" Question: {ques.question}")
print(f" Options: {ques.options}")
print()
answer=MergeConflictAnswer(
question=ques.question,
answer=ques.options[0], # Select first optionmessage="User prefers this option based on recent conversation"
)
answers.append(answer)
# Resolve the conflictconflict.resolve(answers)
print(f"Conflict resolved! Status: {conflict.status}")
print(f"Resolved at: {conflict.resolved_at}")
ifconflict.resolution_data:
print(f"Resolution data: {conflict.resolution_data}")
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptMergeConflictNotFoundErrorase:
print(f"Error: {e}")
exceptMergeConflictAlreadyResolvedErrorase:
print(f"Error: {e}")
exceptMergeConflictInvalidQuestionsErrorase:
print(f"Error: {e}")
ife.invalid_questions:
print(f"Invalid questions: {e.invalid_questions}")
exceptMergeConflictMissingAnswersErrorase:
print(f"Error: {e}")
ife.missing_questions:
print(f"Missing answers for: {e.missing_questions}")
exceptMergeConflictInvalidAnswerErrorase:
print(f"Error: {e}")
ife.questionande.valid_options:
print(f"Question: {e.question}")
print(f"Valid options: {e.valid_options}")
exceptValidationErrorase:
print(f"Error: {e}")

Refresh Merge Conflict Data

fromrecallrai.exceptionsimportUserNotFoundError, MergeConflictNotFoundErrortry:
user=client.get_user("user123")
conflict=user.get_merge_conflict("conflict-uuid")
# Refresh to get latest status from serverconflict.refresh()
print(f"Current status: {conflict.status}")
print(f"Last updated: {conflict.resolved_at}")
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptMergeConflictNotFoundErrorase:
print(f"Error: {e}")

Working with Merge Conflict Statuses

The merge conflict system uses several status values to track the lifecycle of conflicts:

  • PENDING: Conflict detected and waiting for resolution
  • IN_QUEUE: Conflict is queued for automated processing
  • RESOLVING: Conflict is being processed
  • RESOLVED: Conflict has been successfully resolved
  • FAILED: Conflict resolution failed

Example Usage with LLMs

importopenaifromrecallraiimportRecallrAIfromrecallrai.exceptionsimportUserNotFoundErrorfromrecallrai.modelsimportMessageRole# Initialize RecallrAI and OpenAI clientsrai_client=RecallrAI(
api_key="rai_yourapikey", project_id="your-project-uuid"
)
oai_client=openai.OpenAI(api_key="your-openai-api-key")
defchat_with_memory(user_id, session_id=None):
# Get or create usertry:
user=rai_client.get_user(user_id)
exceptUserNotFoundError:
user=rai_client.create_user(user_id)
# Create a new session or get an existing oneifsession_id:
session=user.get_session(session_id=session_id)
else:
session=user.create_session(auto_process_after_seconds=1800)
print(f"Created new session: {session.session_id}")
print("Chat session started. Type 'exit' to end the conversation.")
whileTrue:
# Get user inputuser_message=input("You: ")
ifuser_message.lower() =='exit':
break# Add the user message to RecallrAIsession.add_message(role=MessageRole.USER, content=user_message)
# Get context from RecallrAI after adding the user message# You can specify a recall strategy for different performance/quality trade-offs# Additional parameters like min_top_k, max_top_k, memories_threshold, summaries_threshold, last_n_messages, last_n_summaries are availablecontext=session.get_context() # Uses default BALANCED strategy# Create a system prompt that includes the contextsystem_prompt="You are a helpful assistant"+context.context# Get previous messagesmessages=session.get_messages(offset=0, limit=50)
previous_messages= [{"role": message.role, "content": message.content} formessageinmessages.messages]
# Call the LLM with the system prompt and conversation historyresponse=oai_client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": system_prompt},
**previous_messages,
],
temperature=0.7
)
assistant_message=response.choices[0].message.content# Print the assistant's responseprint(f"Assistant: {assistant_message}")
# Add the assistant's response to RecallrAIsession.add_message(role=MessageRole.ASSISTANT, content=assistant_message)
# Process the session at the end of the conversationprint("Processing session to update memory...")
session.process()
print(f"Session ended. Session ID: {session.session_id}")
returnsession.session_id# Example usageif__name__=="__main__":
user_id="user123"# To continue a previous session, uncomment below and provide the session ID# previous_session_id = "previously-saved-session-uuid"# session_id = chat_with_memory(user_id, previous_session_id)# Start a new sessionsession_id=chat_with_memory(user_id)
print(f"To continue this conversation later, use session ID: {session_id}")

Exception Handling

The RecallrAI SDK implements a comprehensive exception hierarchy to help you handle different error scenarios gracefully:

Base Exception

  • RecallrAIError: The base exception for all SDK-specific errors. All other exceptions inherit from this.

Authentication Errors

  • AuthenticationError: Raised when there's an issue with your API key or project ID authentication.

Network-Related Errors

  • TimeoutError: Occurs when a request takes too long to complete.
  • ConnectionError: Happens when the SDK cannot establish a connection to the RecallrAI API.

Server Errors

  • InternalServerError: Raised when the RecallrAI API returns a 5xx error code.
  • RateLimitError: Raised when the API rate limit has been exceeded (HTTP 429). When available, the retry_after value is provided in the exception details.

User-Related Errors

  • UserNotFoundError: Raised when attempting to access a user that doesn't exist.
  • UserAlreadyExistsError: Occurs when creating a user with an ID that already exists.
  • InvalidCategoriesError: Raised when filtering user memories by categories that do not exist in the project. The exception contains the list of invalid categories in e.invalid_categories.

Session-Related Errors

  • SessionNotFoundError: Raised when attempting to access a non-existent session.
  • InvalidSessionStateError: Occurs when performing an operation that's not valid for the current session state (e.g., adding a message to a processed session).

Merge Conflict-Related Errors

  • MergeConflictError: Base class for merge conflict-related exceptions.
  • MergeConflictNotFoundError: Raised when attempting to access a merge conflict that doesn't exist.
  • MergeConflictAlreadyResolvedError: Occurs when trying to resolve a merge conflict that has already been processed.
  • MergeConflictInvalidQuestionsError: Raised when the provided questions don't match the original clarifying questions.
  • MergeConflictMissingAnswersError: Occurs when not all required clarifying questions have been answered.
  • MergeConflictInvalidAnswerError: Raised when an answer is not one of the valid options for a question.

Input Validation Errors

  • ValidationError: Raised when provided data doesn't meet the required format or constraints.

Importing Exceptions

You can import exceptions directly from the recallrai.exceptions module:

# Import specific exceptionsfromrecallrai.exceptionsimport (
UserNotFoundError, SessionNotFoundError,
InvalidCategoriesError,
MergeConflictNotFoundError,
MergeConflictAlreadyResolvedError,
MergeConflictInvalidQuestionsError,
MergeConflictMissingAnswersError,
MergeConflictInvalidAnswerError,
)
# Import all exceptionsfromrecallrai.exceptionsimport (
RecallrAIError,
AuthenticationError,
TimeoutError,
ConnectionError,
InternalServerError,
RateLimitError,
SessionNotFoundError, InvalidSessionStateError,
UserNotFoundError, UserAlreadyExistsError,
InvalidCategoriesError,
ValidationError,
MergeConflictError,
MergeConflictNotFoundError,
MergeConflictAlreadyResolvedError,
MergeConflictInvalidQuestionsError,
MergeConflictMissingAnswersError,
MergeConflictInvalidAnswerError,
)

Exception Hierarchy Diagram

flowchart LR
%% Title
title[Errors and Exceptions Hierarchy]
style title fill:none,stroke:none
%% Base exception class
Exception[Exception]
RecallrAIError[RecallrAIError]
%% First level exceptions
AuthenticationError[AuthenticationError]
NetworkError[NetworkError]
ServerError[ServerError]
UserError[UserError]
SessionError[SessionError]
MergeConflictError[MergeConflictError]
ValidationError[ValidationError]
%% Second level exceptions - NetworkError children
TimeoutError[TimeoutError]
ConnectionError[ConnectionError]
%% Second level exceptions - ServerError children
InternalServerError[Internal ServerError]
RateLimitError[RateLimitError]
%% Second level exceptions - UserError children
UserNotFoundError[User NotFound Error]
UserAlreadyExistsError[User AlreadyExists Error]
%% Second level exceptions - SessionError children
InvalidSessionStateError[Invalid SessionState Error]
SessionNotFoundError[Session NotFound Error]
%% Second level exceptions - MergeConflictError children
MergeConflictNotFoundError[MergeConflict NotFound Error]
MergeConflictAlreadyResolvedError[MergeConflict AlreadyResolved Error]
MergeConflictInvalidQuestionsError[MergeConflict InvalidQuestions Error]
MergeConflictMissingAnswersError[MergeConflict MissingAnswers Error]
MergeConflictInvalidAnswerError[MergeConflict InvalidAnswer Error]
%% Connect parent to base
Exception --> RecallrAIError
%% Connect base to first level
RecallrAIError --> AuthenticationError
RecallrAIError --> NetworkError
RecallrAIError --> ServerError
RecallrAIError --> UserError
RecallrAIError --> SessionError
RecallrAIError --> MergeConflictError
RecallrAIError --> ValidationError
%% Connect first level to second level
NetworkError --> TimeoutError
NetworkError --> ConnectionError
ServerError --> InternalServerError
ServerError --> RateLimitError
UserError --> UserNotFoundError
UserError --> UserAlreadyExistsError
SessionError --> InvalidSessionStateError
SessionError --> SessionNotFoundError
MergeConflictError --> MergeConflictNotFoundError
MergeConflictError --> MergeConflictAlreadyResolvedError
MergeConflictError --> MergeConflictInvalidQuestionsError
MergeConflictError --> MergeConflictMissingAnswersError
MergeConflictError --> MergeConflictInvalidAnswerError
Loading

Best Practices for Error Handling

When implementing error handling with the RecallrAI SDK, consider these best practices:

  1. Handle specific exceptions first: Catch more specific exceptions before general ones.

    try:
    # SDK operationexceptUserNotFoundError:
    # Specific handlingexceptRecallrAIError:
    # General fallback
  2. Implement retry logic for transient errors: Network and timeout errors might be temporary.

  3. Log detailed error information: Exceptions contain useful information for debugging.

  4. Handle common user flows: For example, check if a user exists before operations, or create them if they don't:

    try:
    user=client.get_user(user_id)
    exceptUserNotFoundError:
    user=client.create_user(user_id)

For more detailed information on specific exceptions, refer to the API documentation.

Conclusion

This README outlines the basic usage of the RecallrAI SDK functions for user and session management. For additional documentation and advanced usage, please see the official documentation or the source code repository on GitHub.

About

Official Python SDK for RecallrAI – a revolutionary contextual memory system that enables AI assistants to form meaningful connections between conversations, just like human memory.

Topics

Resources

Stars

3 stars

Watchers

0 watching

Forks

Releases

Packages

Used by

Contributors

Languages

, 'i'); if (__m === '*' || __re.test(location.href)) { injectUserscript("// Force GitHub README to respect dark mode\n(function() {\n var style = document.createElement('style');\n style.textContent = '\n .markdown-body {\n color-scheme: dark light;\n }\n .markdown-body pre { background: #161b22 !important; }\n .markdown-body code { background: rgba(110, 118, 129, 0.4) !important; }\n .markdown-body table th, .markdown-body table td { border-color: #30363d !important; }\n .markdown-body img { background: #0d1117; }\n .markdown-body blockquote { border-left-color: #8b949e; }\n .markdown-body hr { border-color: #30363d; }\n ';\n document.head.appendChild(style);\n})();", "GitHub Dark Mode README Fix"); } } catch(__e) { console.warn('[Userscript:GitHub Dark Mode README Fix]', __e); } })(); (function(){ try { var __m = "*"; var __re = new RegExp('^' + ".*" + '
Skip to content

Repository files navigation

RecallrAI Python SDK

Official Python SDK for RecallrAI – a revolutionary contextual memory system that enables AI assistants to form meaningful connections between conversations, just like human memory.

Note: All datetime objects returned by the SDK are in UTC timezone.

Installation

Install the SDK via Poetry or pip:

poetry add recallrai
# or
pip install recallrai

Async Support

The SDK provides full async/await support for all operations! Use AsyncRecallrAI, AsyncUser, and AsyncSession for async applications. All usage patterns are identical to the sync versions, just with await keywords.

Initialization

Create a client instance with your API key and project ID:

fromrecallraiimportRecallrAIclient=RecallrAI(
api_key="rai_yourapikey",
project_id="project-uuid",
base_url="https://api.recallrai.com", # custom endpoint if applicabletimeout=60, # seconds
)

User Management

Create a User

fromrecallrai.exceptionsimportUserAlreadyExistsErrortry:
user=client.create_user(user_id="user123", metadata={"name": "John Doe"})
print(f"Created user: {user.user_id}")
print(f"User metadata: {user.metadata}")
print(f"Created at: {user.created_at}")
print(f"Merge conflict enabled: {user.merge_conflict_enabled}") # None = inherit project settingexceptUserAlreadyExistsErrorase:
print(f"Error: {e}")

Get a User

fromrecallrai.exceptionsimportUserNotFoundErrortry:
user=client.get_user("user123")
print(f"User metadata: {user.metadata}")
print(f"Last active: {user.last_active_at}")
exceptUserNotFoundErrorase:
print(f"Error: {e}")

If user_id is already trusted, you can skip the lookup request:

user=client.get_user("user123", validate=False)

List Users

user_list=client.list_users(offset=0, limit=10, metadata_filter={"role": "admin"})
print(f"Total users: {user_list.total}")
print(f"Has more users: {user_list.has_more}")
print("---")
foruinuser_list.users:
print(f"User ID: {u.user_id}")
print(f"Metadata: {u.metadata}")
print(f"Created at: {u.created_at}")
print(f"Last active: {u.last_active_at}")
print("---")

Update a User

fromrecallrai.exceptionsimportUserNotFoundError, UserAlreadyExistsErrortry:
user=client.get_user("user123")
# update() mutates the instance; no value is returneduser.update(
new_metadata={"name": "John Doe", "role": "admin"},
new_user_id="john_doe",
merge_conflict_enabled=True# override: always raise merge conflicts for this user
)
print(f"Updated user ID: {user.user_id}")
print(f"Updated metadata: {user.metadata}")
print(f"Merge conflict enabled: {user.merge_conflict_enabled}")
print(f"Last active: {user.last_active_at}")
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptUserAlreadyExistsErrorase:
print(f"Error: {e}")

Refresh User Instance

fromrecallrai.exceptionsimportUserNotFoundErrortry:
user=client.get_user("john_doe")
user.refresh()
print(f"Refreshed user metadata: {user.metadata}")
print(f"Last active: {user.last_active_at}")
exceptUserNotFoundErrorase:
print(f"Error: {e}")

Delete a User

fromrecallrai.exceptionsimportUserNotFoundErrortry:
user=client.get_user("john_doe")
user.delete()
print("User deleted successfully")
exceptUserNotFoundErrorase:
print(f"Error: {e}")

Session Management

Create a Session

fromrecallrai.exceptionsimportUserNotFoundErrorfromrecallrai.sessionimportSessiontry:
# First, get the useruser=client.get_user("user123")
# Create a session for the user.session: Session=user.create_session(
auto_process_after_seconds=600,
metadata={"type": "chat"}
)
print("Created session id:", session.session_id)
exceptUserNotFoundErrorase:
print(f"Error: {e}")

Get an Existing Session

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundErrortry:
# First, get the useruser=client.get_user("user123")
# Retrieve an existing session by its IDsession=user.get_session(session_id="session-uuid")
print("Session status:", session.status)
print("Session metadata:", session.metadata)
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")

If session_id is trusted, you can skip this lookup request:

session=user.get_session(session_id="session-uuid", validate=False)

Trusted IDs – Skip Validation Lookups

fromrecallrai.modelsimportRecallStrategy# Skips GET /api/v1/users/{user_id}user=client.get_user("user123", validate=False)
# Skips GET /api/v1/users/{user_id}/sessions/{session_id}session=user.get_session(session_id="session-uuid", validate=False)
# Goes directly to context retrievalcontext=session.get_context(recall_strategy=RecallStrategy.LOW_LATENCY)
print(context.context)

Use this only when IDs are already trusted by your system. This optimization skips SDK pre-validation calls.

When validate=False is used, unknown reference fields are set to UNAVAILABLE until you call refresh(). Import it from recallrai.models when you need to check for this sentinel.

Update a Session

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundErrortry:
# First, get the useruser=client.get_user("user123")
# Retrieve an existing session by its IDsession=user.get_session(session_id="session-uuid")
# Update session metadatasession.update(new_metadata={"type": "support_chat"})
print("Updated session metadata:", session.metadata)
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")

Refresh a Session

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundErrortry:
# First, get the useruser=client.get_user("user123")
# Retrieve an existing session by its IDsession=user.get_session(session_id="session-uuid")
# Refresh session data from the serversession.refresh()
print("Session status:", session.status)
print("Refreshed session metadata:", session.metadata)
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")

Delete a Session

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundErrortry:
user=client.get_user("user123")
session=user.get_session(session_id="session-uuid")
session.delete()
print("Session deleted successfully")
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")

List Sessions

fromrecallrai.modelsimportSessionStatusfromrecallrai.exceptionsimportUserNotFoundErrortry:
# First, get the useruser=client.get_user("user123")
# List sessions for this user with optional filterssession_list=user.list_sessions(
offset=0,
limit=10,
metadata_filter={"type": "chat"}, # optional: filter by session metadatastatus_filter=[SessionStatus.PENDING, SessionStatus.PROCESSING] # optional: filter by session status
)
print(f"Total sessions: {session_list.total}")
print(f"Has more sessions: {session_list.has_more}")
forsinsession_list.sessions:
print(s.session_id, s.status, s.metadata)
exceptUserNotFoundErrorase:
print(f"Error: {e}")

Session – Adding Messages

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundError, InvalidSessionStateErrorfromrecallrai.modelsimportMessageRoletry:
# Add a user messagesession.add_message(role=MessageRole.USER, content="Hello! How are you?")
# Add an assistant messagesession.add_message(role=MessageRole.ASSISTANT, content="I'm an assistant. How can I help you?")
# Available message roles:# - MessageRole.USER: Messages from the user/human# - MessageRole.ASSISTANT: Messages from the AI assistantexceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")
exceptInvalidSessionStateErrorase:
print(f"Error: {e}")

Session – Retrieving Context

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundErrorfromrecallrai.modelsimportRecallStrategytry:
# Get context with default parameterscontext=session.get_context()
print("Context:", context.context)
# Get context with specific recall strategycontext=session.get_context(recall_strategy=RecallStrategy.LOW_LATENCY)
print("Context:", context.context)
# Get context with custom memory retrieval parameterscontext=session.get_context(
recall_strategy=RecallStrategy.BALANCED,
min_top_k=10,
max_top_k=100,
memories_threshold=0.6,
summaries_threshold=0.5,
last_n_messages=20,
last_n_summaries=5,
timezone="America/Los_Angeles"# Optional: timezone for timestamp formatting, None for UTC
)
print("Context:", context.context)
# Get context with metadata detailscontext=session.get_context(include_metadata_ids=True)
ifcontext.metadata:
print("Memory IDs:", context.metadata.memory_ids)
print("Session IDs:", context.metadata.session_ids)
print("Vector Queries:", context.metadata.vector_search_queries)
print("Keywords:", context.metadata.keywords)
print("Summary Queries:", context.metadata.session_summaries_search_queries)
print("Date Filters:", context.metadata.date_range_filters)
print("Agent Reasoning:", context.metadata.agent_reasoning)
# Available recall strategies:# - RecallStrategy.LOW_LATENCY: Fast retrieval with basic relevance# - RecallStrategy.BALANCED: Good balance of speed and quality (default)# - RecallStrategy.AGENTIC: Agentic exploration for complex queries# Parameters:# - min_top_k: Minimum number of memories to return (default: 15, range: 5-50)# - max_top_k: Maximum number of memories to return (default: 50, range: 10-100)# - memories_threshold: Similarity threshold for memories (default: 0.6, range: 0.2-0.8)# - summaries_threshold: Similarity threshold for summaries (default: 0.5, range: 0.2-0.8)# - last_n_messages: Number of last messages to include in context (optional, range: 1-100)# - last_n_summaries: Number of last summaries to include in context (optional, range: 1-20)# - timezone: Timezone for formatting timestamps (optional, e.g., 'America/New_York', None for UTC)# - include_system_prompt: Whether to include the default system prompt of Recallr AI (default: True)exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")

Session – Streaming Context

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundErrorfromrecallrai.modelsimportRecallStrategytry:
foreventinsession.get_context_stream(
recall_strategy=RecallStrategy.BALANCED,
timezone="America/Los_Angeles",
):
ifevent.status_update_message:
print("Status:", event.status_update_message)
ifevent.metadata:
print("Metadata:", event.metadata)
ifevent.is_final:
ifevent.error_message:
print("Error:", event.error_message)
else:
print("Final context:", event.context)
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")

Session – Process Session

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundError, InvalidSessionStateErrortry:
session.process()
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")
exceptInvalidSessionStateErrorase:
print(f"Error: {e}")

Session – List Messages

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundErrortry:
# Paginated retrievalmessages=session.get_messages(offset=0, limit=50)
formsginmessages.messages:
print(f"{msg.role.value.upper()} (at {msg.timestamp}): {msg.content}")
print(f"Has more?: {messages.has_more}")
print(f"Total messages: {messages.total}")
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")

User Memories

List User Memories (with optional category filters)

fromrecallrai.exceptionsimportUserNotFoundError, InvalidCategoriesErrortry:
user=client.get_user("user123")
# List memories with all available optionsmemories=user.list_memories(
categories=["food_preferences", "allergies"], # optional: filter by categoriessession_id_filter=["session-uuid-1", "session-uuid-2"], # optional: filter by specific sessionssession_metadata_filter={"environment": "production"}, # optional: filter by session metadataoffset=0,
limit=20, # max 200include_previous_versions=True, # default: True - include version historyinclude_connected_memories=True, # default: True - include related memories
)
formeminmemories.items:
print(f"Memory ID: {mem.memory_id}")
print(f"Categories: {mem.categories}")
print(f"Content: {mem.content}")
print(f"Created at: {mem.created_at}")
print(f"Session ID: {mem.session_id}")
# Version informationprint(f"Version: {mem.version_number} of {mem.total_versions}")
print(f"Has previous versions: {mem.has_previous_versions}")
# Previous versions (if included)ifmem.previous_versions:
print(f"Previous versions: {len(mem.previous_versions)}")
forversioninmem.previous_versions:
print(f" - Version {version.version_number}: {version.content}")
print(f" Created: {version.created_at}, Expired: {version.expired_at}")
print(f" Expiration reason: {version.expiration_reason}")
# Connected memories (if included)ifmem.connected_memories:
print(f"Connected memories: {len(mem.connected_memories)}")
forconnectedinmem.connected_memories:
print(f" - {connected.memory_id}: {connected.content}")
# Merge conflict statusprint(f"Merge conflict in progress: {mem.merge_conflict_in_progress}")
print("---")
print(f"Has more?: {memories.has_more}")
print(f"Total memories: {memories.total}")
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptInvalidCategoriesErrorase:
print(f"Invalid categories: {e.invalid_categories}")
print(f"Error: {e}")

Memory Item Fields

Each memory item returned contains the following information:

  • memory_id: Unique identifier for the current/latest version of the memory
  • categories: List of category strings the memory belongs to
  • content: The current version's content text
  • created_at: Timestamp when the latest version was created
  • session_id: ID of the session that created this version
  • version_number: Which version this is (e.g., 3 means this is the 3rd version)
  • total_versions: Total number of versions that exist for this memory
  • has_previous_versions: Boolean indicating if total_versions > 1
  • previous_versions (optional): List of MemoryVersionInfo objects containing:
    • version_number: Sequential version number (1 = oldest)
    • content: Content of that version
    • created_at: When this version was created
    • expired_at: When this version expired
    • expiration_reason: Why it expired (e.g., new version created)
  • connected_memories (optional): List of MemoryRelationship objects containing:
    • memory_id: ID of the connected memory
    • content: Brief content for context
  • merge_conflict_in_progress: Boolean indicating if this memory has an active merge conflict

User Messages

Get Last N Messages

Retrieve the most recent messages for a user across all their sessions. This is particularly useful for chatbot applications where you need conversation context, such as WhatsApp support bots where you want to pass the last few messages to understand the ongoing conversation.

fromrecallrai.exceptionsimportUserNotFoundErrortry:
user=client.get_user("user123")
# Fetch last N messages (e.g., last 5)messages=user.get_last_n_messages(n=5)
formsginmessages.messages:
print(f"Session ID: {msg.session_id}")
print(f"{msg.role.upper()} (at {msg.timestamp}): {msg.content}")
print("---")
exceptUserNotFoundErrorase:
print(f"Error: {e}")

Merge Conflict Management

When RecallrAI processes sessions, it may detect conflicts between new memories and existing ones. The SDK provides comprehensive tools to handle these merge conflicts, allowing you to guide the resolution process through clarifying questions.

List Merge Conflicts

fromrecallrai.exceptionsimportUserNotFoundErrorfromrecallrai.modelsimportMergeConflictStatustry:
user=client.get_user("user123")
# List all merge conflicts (with optional status filter)conflicts=user.list_merge_conflicts(
offset=0,
limit=10,
status=MergeConflictStatus.PENDING, # optional: filter by statussort_by="created_at", # created_at, resolved_atsort_order="desc", # asc, desc
)
print(f"Total conflicts: {conflicts.total}")
print(f"Has more: {conflicts.has_more}")
forconfinconflicts.conflicts:
print(f"Conflict ID: {conf.conflict_id}")
print(f"Status: {conf.status}")
print(f"New memory: {conf.proposed_memory_content}")
print(f"Conflicting memories: {len(conf.conflicting_memories)}")
print(f"Questions: {len(conf.clarifying_questions)}")
print(f"Created at: {conf.created_at}")
print("---")
exceptUserNotFoundErrorase:
print(f"Error: {e}")

Get a Specific Merge Conflict

fromrecallrai.exceptionsimportUserNotFoundError, MergeConflictNotFoundErrortry:
user=client.get_user("user123")
conflict=user.get_merge_conflict("conflict-uuid")
print(f"Conflict ID: {conflict.conflict_id}")
print(f"Status: {conflict.status.value}")
print(f"New memory content: {conflict.proposed_memory_content}")
# Examine conflicting memoriesprint("\nConflicting memories:")
formeminconflict.conflicting_memories:
print(f" Content: {mem.content}")
print(f" Reason: {mem.reason}")
print()
# View clarifying questionsprint("Clarifying questions:")
forquesinconflict.clarifying_questions:
print(f" Question: {ques.question}")
print(f" Options: {ques.options}")
print()
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptMergeConflictNotFoundErrorase:
print(f"Error: {e}")

Resolve a Merge Conflict

fromrecallrai.exceptionsimport (
UserNotFoundError, MergeConflictNotFoundError,
MergeConflictAlreadyResolvedError,
MergeConflictInvalidQuestionsError,
MergeConflictMissingAnswersError,
MergeConflictInvalidAnswerError,
ValidationError
)
fromrecallrai.modelsimportMergeConflictAnswertry:
user=client.get_user("user123")
conflict=user.get_merge_conflict("conflict-uuid")
# Prepare answers to the clarifying questionsanswers= []
forquesinconflict.clarifying_questions:
print(f" Question: {ques.question}")
print(f" Options: {ques.options}")
print()
answer=MergeConflictAnswer(
question=ques.question,
answer=ques.options[0], # Select first optionmessage="User prefers this option based on recent conversation"
)
answers.append(answer)
# Resolve the conflictconflict.resolve(answers)
print(f"Conflict resolved! Status: {conflict.status}")
print(f"Resolved at: {conflict.resolved_at}")
ifconflict.resolution_data:
print(f"Resolution data: {conflict.resolution_data}")
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptMergeConflictNotFoundErrorase:
print(f"Error: {e}")
exceptMergeConflictAlreadyResolvedErrorase:
print(f"Error: {e}")
exceptMergeConflictInvalidQuestionsErrorase:
print(f"Error: {e}")
ife.invalid_questions:
print(f"Invalid questions: {e.invalid_questions}")
exceptMergeConflictMissingAnswersErrorase:
print(f"Error: {e}")
ife.missing_questions:
print(f"Missing answers for: {e.missing_questions}")
exceptMergeConflictInvalidAnswerErrorase:
print(f"Error: {e}")
ife.questionande.valid_options:
print(f"Question: {e.question}")
print(f"Valid options: {e.valid_options}")
exceptValidationErrorase:
print(f"Error: {e}")

Refresh Merge Conflict Data

fromrecallrai.exceptionsimportUserNotFoundError, MergeConflictNotFoundErrortry:
user=client.get_user("user123")
conflict=user.get_merge_conflict("conflict-uuid")
# Refresh to get latest status from serverconflict.refresh()
print(f"Current status: {conflict.status}")
print(f"Last updated: {conflict.resolved_at}")
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptMergeConflictNotFoundErrorase:
print(f"Error: {e}")

Working with Merge Conflict Statuses

The merge conflict system uses several status values to track the lifecycle of conflicts:

  • PENDING: Conflict detected and waiting for resolution
  • IN_QUEUE: Conflict is queued for automated processing
  • RESOLVING: Conflict is being processed
  • RESOLVED: Conflict has been successfully resolved
  • FAILED: Conflict resolution failed

Example Usage with LLMs

importopenaifromrecallraiimportRecallrAIfromrecallrai.exceptionsimportUserNotFoundErrorfromrecallrai.modelsimportMessageRole# Initialize RecallrAI and OpenAI clientsrai_client=RecallrAI(
api_key="rai_yourapikey", project_id="your-project-uuid"
)
oai_client=openai.OpenAI(api_key="your-openai-api-key")
defchat_with_memory(user_id, session_id=None):
# Get or create usertry:
user=rai_client.get_user(user_id)
exceptUserNotFoundError:
user=rai_client.create_user(user_id)
# Create a new session or get an existing oneifsession_id:
session=user.get_session(session_id=session_id)
else:
session=user.create_session(auto_process_after_seconds=1800)
print(f"Created new session: {session.session_id}")
print("Chat session started. Type 'exit' to end the conversation.")
whileTrue:
# Get user inputuser_message=input("You: ")
ifuser_message.lower() =='exit':
break# Add the user message to RecallrAIsession.add_message(role=MessageRole.USER, content=user_message)
# Get context from RecallrAI after adding the user message# You can specify a recall strategy for different performance/quality trade-offs# Additional parameters like min_top_k, max_top_k, memories_threshold, summaries_threshold, last_n_messages, last_n_summaries are availablecontext=session.get_context() # Uses default BALANCED strategy# Create a system prompt that includes the contextsystem_prompt="You are a helpful assistant"+context.context# Get previous messagesmessages=session.get_messages(offset=0, limit=50)
previous_messages= [{"role": message.role, "content": message.content} formessageinmessages.messages]
# Call the LLM with the system prompt and conversation historyresponse=oai_client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": system_prompt},
**previous_messages,
],
temperature=0.7
)
assistant_message=response.choices[0].message.content# Print the assistant's responseprint(f"Assistant: {assistant_message}")
# Add the assistant's response to RecallrAIsession.add_message(role=MessageRole.ASSISTANT, content=assistant_message)
# Process the session at the end of the conversationprint("Processing session to update memory...")
session.process()
print(f"Session ended. Session ID: {session.session_id}")
returnsession.session_id# Example usageif__name__=="__main__":
user_id="user123"# To continue a previous session, uncomment below and provide the session ID# previous_session_id = "previously-saved-session-uuid"# session_id = chat_with_memory(user_id, previous_session_id)# Start a new sessionsession_id=chat_with_memory(user_id)
print(f"To continue this conversation later, use session ID: {session_id}")

Exception Handling

The RecallrAI SDK implements a comprehensive exception hierarchy to help you handle different error scenarios gracefully:

Base Exception

  • RecallrAIError: The base exception for all SDK-specific errors. All other exceptions inherit from this.

Authentication Errors

  • AuthenticationError: Raised when there's an issue with your API key or project ID authentication.

Network-Related Errors

  • TimeoutError: Occurs when a request takes too long to complete.
  • ConnectionError: Happens when the SDK cannot establish a connection to the RecallrAI API.

Server Errors

  • InternalServerError: Raised when the RecallrAI API returns a 5xx error code.
  • RateLimitError: Raised when the API rate limit has been exceeded (HTTP 429). When available, the retry_after value is provided in the exception details.

User-Related Errors

  • UserNotFoundError: Raised when attempting to access a user that doesn't exist.
  • UserAlreadyExistsError: Occurs when creating a user with an ID that already exists.
  • InvalidCategoriesError: Raised when filtering user memories by categories that do not exist in the project. The exception contains the list of invalid categories in e.invalid_categories.

Session-Related Errors

  • SessionNotFoundError: Raised when attempting to access a non-existent session.
  • InvalidSessionStateError: Occurs when performing an operation that's not valid for the current session state (e.g., adding a message to a processed session).

Merge Conflict-Related Errors

  • MergeConflictError: Base class for merge conflict-related exceptions.
  • MergeConflictNotFoundError: Raised when attempting to access a merge conflict that doesn't exist.
  • MergeConflictAlreadyResolvedError: Occurs when trying to resolve a merge conflict that has already been processed.
  • MergeConflictInvalidQuestionsError: Raised when the provided questions don't match the original clarifying questions.
  • MergeConflictMissingAnswersError: Occurs when not all required clarifying questions have been answered.
  • MergeConflictInvalidAnswerError: Raised when an answer is not one of the valid options for a question.

Input Validation Errors

  • ValidationError: Raised when provided data doesn't meet the required format or constraints.

Importing Exceptions

You can import exceptions directly from the recallrai.exceptions module:

# Import specific exceptionsfromrecallrai.exceptionsimport (
UserNotFoundError, SessionNotFoundError,
InvalidCategoriesError,
MergeConflictNotFoundError,
MergeConflictAlreadyResolvedError,
MergeConflictInvalidQuestionsError,
MergeConflictMissingAnswersError,
MergeConflictInvalidAnswerError,
)
# Import all exceptionsfromrecallrai.exceptionsimport (
RecallrAIError,
AuthenticationError,
TimeoutError,
ConnectionError,
InternalServerError,
RateLimitError,
SessionNotFoundError, InvalidSessionStateError,
UserNotFoundError, UserAlreadyExistsError,
InvalidCategoriesError,
ValidationError,
MergeConflictError,
MergeConflictNotFoundError,
MergeConflictAlreadyResolvedError,
MergeConflictInvalidQuestionsError,
MergeConflictMissingAnswersError,
MergeConflictInvalidAnswerError,
)

Exception Hierarchy Diagram

flowchart LR
%% Title
title[Errors and Exceptions Hierarchy]
style title fill:none,stroke:none
%% Base exception class
Exception[Exception]
RecallrAIError[RecallrAIError]
%% First level exceptions
AuthenticationError[AuthenticationError]
NetworkError[NetworkError]
ServerError[ServerError]
UserError[UserError]
SessionError[SessionError]
MergeConflictError[MergeConflictError]
ValidationError[ValidationError]
%% Second level exceptions - NetworkError children
TimeoutError[TimeoutError]
ConnectionError[ConnectionError]
%% Second level exceptions - ServerError children
InternalServerError[Internal ServerError]
RateLimitError[RateLimitError]
%% Second level exceptions - UserError children
UserNotFoundError[User NotFound Error]
UserAlreadyExistsError[User AlreadyExists Error]
%% Second level exceptions - SessionError children
InvalidSessionStateError[Invalid SessionState Error]
SessionNotFoundError[Session NotFound Error]
%% Second level exceptions - MergeConflictError children
MergeConflictNotFoundError[MergeConflict NotFound Error]
MergeConflictAlreadyResolvedError[MergeConflict AlreadyResolved Error]
MergeConflictInvalidQuestionsError[MergeConflict InvalidQuestions Error]
MergeConflictMissingAnswersError[MergeConflict MissingAnswers Error]
MergeConflictInvalidAnswerError[MergeConflict InvalidAnswer Error]
%% Connect parent to base
Exception --> RecallrAIError
%% Connect base to first level
RecallrAIError --> AuthenticationError
RecallrAIError --> NetworkError
RecallrAIError --> ServerError
RecallrAIError --> UserError
RecallrAIError --> SessionError
RecallrAIError --> MergeConflictError
RecallrAIError --> ValidationError
%% Connect first level to second level
NetworkError --> TimeoutError
NetworkError --> ConnectionError
ServerError --> InternalServerError
ServerError --> RateLimitError
UserError --> UserNotFoundError
UserError --> UserAlreadyExistsError
SessionError --> InvalidSessionStateError
SessionError --> SessionNotFoundError
MergeConflictError --> MergeConflictNotFoundError
MergeConflictError --> MergeConflictAlreadyResolvedError
MergeConflictError --> MergeConflictInvalidQuestionsError
MergeConflictError --> MergeConflictMissingAnswersError
MergeConflictError --> MergeConflictInvalidAnswerError
Loading

Best Practices for Error Handling

When implementing error handling with the RecallrAI SDK, consider these best practices:

  1. Handle specific exceptions first: Catch more specific exceptions before general ones.

    try:
    # SDK operationexceptUserNotFoundError:
    # Specific handlingexceptRecallrAIError:
    # General fallback
  2. Implement retry logic for transient errors: Network and timeout errors might be temporary.

  3. Log detailed error information: Exceptions contain useful information for debugging.

  4. Handle common user flows: For example, check if a user exists before operations, or create them if they don't:

    try:
    user=client.get_user(user_id)
    exceptUserNotFoundError:
    user=client.create_user(user_id)

For more detailed information on specific exceptions, refer to the API documentation.

Conclusion

This README outlines the basic usage of the RecallrAI SDK functions for user and session management. For additional documentation and advanced usage, please see the official documentation or the source code repository on GitHub.

About

Official Python SDK for RecallrAI – a revolutionary contextual memory system that enables AI assistants to form meaningful connections between conversations, just like human memory.

Topics

Resources

Stars

3 stars

Watchers

0 watching

Forks

Releases

Packages

Used by

Contributors

Languages

, 'i'); if (__m === '*' || __re.test(location.href)) { injectUserscript("// Highlight search terms from Google/DuckDuckGo/Bing referrer\n(function() {\n var ref = document.referrer;\n var terms = [];\n \n if (ref.includes('google.com') || ref.includes('duckduckgo.com') || ref.includes('bing.com')) {\n var url = new URL(ref);\n var q = url.searchParams.get('q') || url.searchParams.get('p');\n if (q) {\n terms = q.split(/\\s+/).filter(function(t) { return t.length \u003e 2; });\n }\n }\n \n if (terms.length === 0) return;\n \n var style = document.createElement('style');\n style.textContent = '.userscript-highlight { background: #fbbf24; color: #1a1a2e; padding: 1px 3px; border-radius: 2px; }';\n document.head.appendChild(style);\n \n function highlight(node) {\n if (node.nodeType === 3) { // text node\n var text = node.textContent;\n var found = false;\n terms.forEach(function(term) {\n var regex = new RegExp('(' + term.replace(/[.*+?^${}()|[\\]\\\\]/g, '\\\\') + ')', 'gi');\n if (regex.test(text)) {\n found = true;\n var frag = document.createDocumentFragment();\n var parts = text.split(regex);\n parts.forEach(function(part, i) {\n if (i % 2 === 0) {\n frag.appendChild(document.createTextNode(part));\n } else {\n var span = document.createElement('span');\n span.className = 'userscript-highlight';\n span.textContent = part;\n frag.appendChild(span);\n }\n });\n node.parentNode.replaceChild(frag, node);\n }\n });\n } else if (node.nodeType === 1 && node.childNodes) { // element\n var skipTags = ['SCRIPT', 'STYLE', 'NOSCRIPT', 'TEXTAREA', 'INPUT', 'SELECT'];\n if (!skipTags.includes(node.tagName)) {\n Array.from(node.childNodes).forEach(highlight);\n }\n }\n }\n \n highlight(document.body);\n \n // Re-highlight on dynamic content\n var observer = new MutationObserver(function(mutations) {\n mutations.forEach(function(m) {\n m.addedNodes.forEach(function(node) {\n if (node.nodeType === 1 || node.nodeType === 3) highlight(node);\n });\n });\n });\n observer.observe(document.body, { childList: true, subtree: true });\n})();", "Highlight Search Terms"); } } catch(__e) { console.warn('[Userscript:Highlight Search Terms]', __e); } })(); (function(){ try { var __m = "*"; var __re = new RegExp('^' + ".*" + '
Skip to content

Repository files navigation

RecallrAI Python SDK

Official Python SDK for RecallrAI – a revolutionary contextual memory system that enables AI assistants to form meaningful connections between conversations, just like human memory.

Note: All datetime objects returned by the SDK are in UTC timezone.

Installation

Install the SDK via Poetry or pip:

poetry add recallrai
# or
pip install recallrai

Async Support

The SDK provides full async/await support for all operations! Use AsyncRecallrAI, AsyncUser, and AsyncSession for async applications. All usage patterns are identical to the sync versions, just with await keywords.

Initialization

Create a client instance with your API key and project ID:

fromrecallraiimportRecallrAIclient=RecallrAI(
api_key="rai_yourapikey",
project_id="project-uuid",
base_url="https://api.recallrai.com", # custom endpoint if applicabletimeout=60, # seconds
)

User Management

Create a User

fromrecallrai.exceptionsimportUserAlreadyExistsErrortry:
user=client.create_user(user_id="user123", metadata={"name": "John Doe"})
print(f"Created user: {user.user_id}")
print(f"User metadata: {user.metadata}")
print(f"Created at: {user.created_at}")
print(f"Merge conflict enabled: {user.merge_conflict_enabled}") # None = inherit project settingexceptUserAlreadyExistsErrorase:
print(f"Error: {e}")

Get a User

fromrecallrai.exceptionsimportUserNotFoundErrortry:
user=client.get_user("user123")
print(f"User metadata: {user.metadata}")
print(f"Last active: {user.last_active_at}")
exceptUserNotFoundErrorase:
print(f"Error: {e}")

If user_id is already trusted, you can skip the lookup request:

user=client.get_user("user123", validate=False)

List Users

user_list=client.list_users(offset=0, limit=10, metadata_filter={"role": "admin"})
print(f"Total users: {user_list.total}")
print(f"Has more users: {user_list.has_more}")
print("---")
foruinuser_list.users:
print(f"User ID: {u.user_id}")
print(f"Metadata: {u.metadata}")
print(f"Created at: {u.created_at}")
print(f"Last active: {u.last_active_at}")
print("---")

Update a User

fromrecallrai.exceptionsimportUserNotFoundError, UserAlreadyExistsErrortry:
user=client.get_user("user123")
# update() mutates the instance; no value is returneduser.update(
new_metadata={"name": "John Doe", "role": "admin"},
new_user_id="john_doe",
merge_conflict_enabled=True# override: always raise merge conflicts for this user
)
print(f"Updated user ID: {user.user_id}")
print(f"Updated metadata: {user.metadata}")
print(f"Merge conflict enabled: {user.merge_conflict_enabled}")
print(f"Last active: {user.last_active_at}")
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptUserAlreadyExistsErrorase:
print(f"Error: {e}")

Refresh User Instance

fromrecallrai.exceptionsimportUserNotFoundErrortry:
user=client.get_user("john_doe")
user.refresh()
print(f"Refreshed user metadata: {user.metadata}")
print(f"Last active: {user.last_active_at}")
exceptUserNotFoundErrorase:
print(f"Error: {e}")

Delete a User

fromrecallrai.exceptionsimportUserNotFoundErrortry:
user=client.get_user("john_doe")
user.delete()
print("User deleted successfully")
exceptUserNotFoundErrorase:
print(f"Error: {e}")

Session Management

Create a Session

fromrecallrai.exceptionsimportUserNotFoundErrorfromrecallrai.sessionimportSessiontry:
# First, get the useruser=client.get_user("user123")
# Create a session for the user.session: Session=user.create_session(
auto_process_after_seconds=600,
metadata={"type": "chat"}
)
print("Created session id:", session.session_id)
exceptUserNotFoundErrorase:
print(f"Error: {e}")

Get an Existing Session

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundErrortry:
# First, get the useruser=client.get_user("user123")
# Retrieve an existing session by its IDsession=user.get_session(session_id="session-uuid")
print("Session status:", session.status)
print("Session metadata:", session.metadata)
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")

If session_id is trusted, you can skip this lookup request:

session=user.get_session(session_id="session-uuid", validate=False)

Trusted IDs – Skip Validation Lookups

fromrecallrai.modelsimportRecallStrategy# Skips GET /api/v1/users/{user_id}user=client.get_user("user123", validate=False)
# Skips GET /api/v1/users/{user_id}/sessions/{session_id}session=user.get_session(session_id="session-uuid", validate=False)
# Goes directly to context retrievalcontext=session.get_context(recall_strategy=RecallStrategy.LOW_LATENCY)
print(context.context)

Use this only when IDs are already trusted by your system. This optimization skips SDK pre-validation calls.

When validate=False is used, unknown reference fields are set to UNAVAILABLE until you call refresh(). Import it from recallrai.models when you need to check for this sentinel.

Update a Session

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundErrortry:
# First, get the useruser=client.get_user("user123")
# Retrieve an existing session by its IDsession=user.get_session(session_id="session-uuid")
# Update session metadatasession.update(new_metadata={"type": "support_chat"})
print("Updated session metadata:", session.metadata)
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")

Refresh a Session

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundErrortry:
# First, get the useruser=client.get_user("user123")
# Retrieve an existing session by its IDsession=user.get_session(session_id="session-uuid")
# Refresh session data from the serversession.refresh()
print("Session status:", session.status)
print("Refreshed session metadata:", session.metadata)
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")

Delete a Session

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundErrortry:
user=client.get_user("user123")
session=user.get_session(session_id="session-uuid")
session.delete()
print("Session deleted successfully")
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")

List Sessions

fromrecallrai.modelsimportSessionStatusfromrecallrai.exceptionsimportUserNotFoundErrortry:
# First, get the useruser=client.get_user("user123")
# List sessions for this user with optional filterssession_list=user.list_sessions(
offset=0,
limit=10,
metadata_filter={"type": "chat"}, # optional: filter by session metadatastatus_filter=[SessionStatus.PENDING, SessionStatus.PROCESSING] # optional: filter by session status
)
print(f"Total sessions: {session_list.total}")
print(f"Has more sessions: {session_list.has_more}")
forsinsession_list.sessions:
print(s.session_id, s.status, s.metadata)
exceptUserNotFoundErrorase:
print(f"Error: {e}")

Session – Adding Messages

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundError, InvalidSessionStateErrorfromrecallrai.modelsimportMessageRoletry:
# Add a user messagesession.add_message(role=MessageRole.USER, content="Hello! How are you?")
# Add an assistant messagesession.add_message(role=MessageRole.ASSISTANT, content="I'm an assistant. How can I help you?")
# Available message roles:# - MessageRole.USER: Messages from the user/human# - MessageRole.ASSISTANT: Messages from the AI assistantexceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")
exceptInvalidSessionStateErrorase:
print(f"Error: {e}")

Session – Retrieving Context

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundErrorfromrecallrai.modelsimportRecallStrategytry:
# Get context with default parameterscontext=session.get_context()
print("Context:", context.context)
# Get context with specific recall strategycontext=session.get_context(recall_strategy=RecallStrategy.LOW_LATENCY)
print("Context:", context.context)
# Get context with custom memory retrieval parameterscontext=session.get_context(
recall_strategy=RecallStrategy.BALANCED,
min_top_k=10,
max_top_k=100,
memories_threshold=0.6,
summaries_threshold=0.5,
last_n_messages=20,
last_n_summaries=5,
timezone="America/Los_Angeles"# Optional: timezone for timestamp formatting, None for UTC
)
print("Context:", context.context)
# Get context with metadata detailscontext=session.get_context(include_metadata_ids=True)
ifcontext.metadata:
print("Memory IDs:", context.metadata.memory_ids)
print("Session IDs:", context.metadata.session_ids)
print("Vector Queries:", context.metadata.vector_search_queries)
print("Keywords:", context.metadata.keywords)
print("Summary Queries:", context.metadata.session_summaries_search_queries)
print("Date Filters:", context.metadata.date_range_filters)
print("Agent Reasoning:", context.metadata.agent_reasoning)
# Available recall strategies:# - RecallStrategy.LOW_LATENCY: Fast retrieval with basic relevance# - RecallStrategy.BALANCED: Good balance of speed and quality (default)# - RecallStrategy.AGENTIC: Agentic exploration for complex queries# Parameters:# - min_top_k: Minimum number of memories to return (default: 15, range: 5-50)# - max_top_k: Maximum number of memories to return (default: 50, range: 10-100)# - memories_threshold: Similarity threshold for memories (default: 0.6, range: 0.2-0.8)# - summaries_threshold: Similarity threshold for summaries (default: 0.5, range: 0.2-0.8)# - last_n_messages: Number of last messages to include in context (optional, range: 1-100)# - last_n_summaries: Number of last summaries to include in context (optional, range: 1-20)# - timezone: Timezone for formatting timestamps (optional, e.g., 'America/New_York', None for UTC)# - include_system_prompt: Whether to include the default system prompt of Recallr AI (default: True)exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")

Session – Streaming Context

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundErrorfromrecallrai.modelsimportRecallStrategytry:
foreventinsession.get_context_stream(
recall_strategy=RecallStrategy.BALANCED,
timezone="America/Los_Angeles",
):
ifevent.status_update_message:
print("Status:", event.status_update_message)
ifevent.metadata:
print("Metadata:", event.metadata)
ifevent.is_final:
ifevent.error_message:
print("Error:", event.error_message)
else:
print("Final context:", event.context)
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")

Session – Process Session

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundError, InvalidSessionStateErrortry:
session.process()
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")
exceptInvalidSessionStateErrorase:
print(f"Error: {e}")

Session – List Messages

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundErrortry:
# Paginated retrievalmessages=session.get_messages(offset=0, limit=50)
formsginmessages.messages:
print(f"{msg.role.value.upper()} (at {msg.timestamp}): {msg.content}")
print(f"Has more?: {messages.has_more}")
print(f"Total messages: {messages.total}")
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")

User Memories

List User Memories (with optional category filters)

fromrecallrai.exceptionsimportUserNotFoundError, InvalidCategoriesErrortry:
user=client.get_user("user123")
# List memories with all available optionsmemories=user.list_memories(
categories=["food_preferences", "allergies"], # optional: filter by categoriessession_id_filter=["session-uuid-1", "session-uuid-2"], # optional: filter by specific sessionssession_metadata_filter={"environment": "production"}, # optional: filter by session metadataoffset=0,
limit=20, # max 200include_previous_versions=True, # default: True - include version historyinclude_connected_memories=True, # default: True - include related memories
)
formeminmemories.items:
print(f"Memory ID: {mem.memory_id}")
print(f"Categories: {mem.categories}")
print(f"Content: {mem.content}")
print(f"Created at: {mem.created_at}")
print(f"Session ID: {mem.session_id}")
# Version informationprint(f"Version: {mem.version_number} of {mem.total_versions}")
print(f"Has previous versions: {mem.has_previous_versions}")
# Previous versions (if included)ifmem.previous_versions:
print(f"Previous versions: {len(mem.previous_versions)}")
forversioninmem.previous_versions:
print(f" - Version {version.version_number}: {version.content}")
print(f" Created: {version.created_at}, Expired: {version.expired_at}")
print(f" Expiration reason: {version.expiration_reason}")
# Connected memories (if included)ifmem.connected_memories:
print(f"Connected memories: {len(mem.connected_memories)}")
forconnectedinmem.connected_memories:
print(f" - {connected.memory_id}: {connected.content}")
# Merge conflict statusprint(f"Merge conflict in progress: {mem.merge_conflict_in_progress}")
print("---")
print(f"Has more?: {memories.has_more}")
print(f"Total memories: {memories.total}")
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptInvalidCategoriesErrorase:
print(f"Invalid categories: {e.invalid_categories}")
print(f"Error: {e}")

Memory Item Fields

Each memory item returned contains the following information:

  • memory_id: Unique identifier for the current/latest version of the memory
  • categories: List of category strings the memory belongs to
  • content: The current version's content text
  • created_at: Timestamp when the latest version was created
  • session_id: ID of the session that created this version
  • version_number: Which version this is (e.g., 3 means this is the 3rd version)
  • total_versions: Total number of versions that exist for this memory
  • has_previous_versions: Boolean indicating if total_versions > 1
  • previous_versions (optional): List of MemoryVersionInfo objects containing:
    • version_number: Sequential version number (1 = oldest)
    • content: Content of that version
    • created_at: When this version was created
    • expired_at: When this version expired
    • expiration_reason: Why it expired (e.g., new version created)
  • connected_memories (optional): List of MemoryRelationship objects containing:
    • memory_id: ID of the connected memory
    • content: Brief content for context
  • merge_conflict_in_progress: Boolean indicating if this memory has an active merge conflict

User Messages

Get Last N Messages

Retrieve the most recent messages for a user across all their sessions. This is particularly useful for chatbot applications where you need conversation context, such as WhatsApp support bots where you want to pass the last few messages to understand the ongoing conversation.

fromrecallrai.exceptionsimportUserNotFoundErrortry:
user=client.get_user("user123")
# Fetch last N messages (e.g., last 5)messages=user.get_last_n_messages(n=5)
formsginmessages.messages:
print(f"Session ID: {msg.session_id}")
print(f"{msg.role.upper()} (at {msg.timestamp}): {msg.content}")
print("---")
exceptUserNotFoundErrorase:
print(f"Error: {e}")

Merge Conflict Management

When RecallrAI processes sessions, it may detect conflicts between new memories and existing ones. The SDK provides comprehensive tools to handle these merge conflicts, allowing you to guide the resolution process through clarifying questions.

List Merge Conflicts

fromrecallrai.exceptionsimportUserNotFoundErrorfromrecallrai.modelsimportMergeConflictStatustry:
user=client.get_user("user123")
# List all merge conflicts (with optional status filter)conflicts=user.list_merge_conflicts(
offset=0,
limit=10,
status=MergeConflictStatus.PENDING, # optional: filter by statussort_by="created_at", # created_at, resolved_atsort_order="desc", # asc, desc
)
print(f"Total conflicts: {conflicts.total}")
print(f"Has more: {conflicts.has_more}")
forconfinconflicts.conflicts:
print(f"Conflict ID: {conf.conflict_id}")
print(f"Status: {conf.status}")
print(f"New memory: {conf.proposed_memory_content}")
print(f"Conflicting memories: {len(conf.conflicting_memories)}")
print(f"Questions: {len(conf.clarifying_questions)}")
print(f"Created at: {conf.created_at}")
print("---")
exceptUserNotFoundErrorase:
print(f"Error: {e}")

Get a Specific Merge Conflict

fromrecallrai.exceptionsimportUserNotFoundError, MergeConflictNotFoundErrortry:
user=client.get_user("user123")
conflict=user.get_merge_conflict("conflict-uuid")
print(f"Conflict ID: {conflict.conflict_id}")
print(f"Status: {conflict.status.value}")
print(f"New memory content: {conflict.proposed_memory_content}")
# Examine conflicting memoriesprint("\nConflicting memories:")
formeminconflict.conflicting_memories:
print(f" Content: {mem.content}")
print(f" Reason: {mem.reason}")
print()
# View clarifying questionsprint("Clarifying questions:")
forquesinconflict.clarifying_questions:
print(f" Question: {ques.question}")
print(f" Options: {ques.options}")
print()
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptMergeConflictNotFoundErrorase:
print(f"Error: {e}")

Resolve a Merge Conflict

fromrecallrai.exceptionsimport (
UserNotFoundError, MergeConflictNotFoundError,
MergeConflictAlreadyResolvedError,
MergeConflictInvalidQuestionsError,
MergeConflictMissingAnswersError,
MergeConflictInvalidAnswerError,
ValidationError
)
fromrecallrai.modelsimportMergeConflictAnswertry:
user=client.get_user("user123")
conflict=user.get_merge_conflict("conflict-uuid")
# Prepare answers to the clarifying questionsanswers= []
forquesinconflict.clarifying_questions:
print(f" Question: {ques.question}")
print(f" Options: {ques.options}")
print()
answer=MergeConflictAnswer(
question=ques.question,
answer=ques.options[0], # Select first optionmessage="User prefers this option based on recent conversation"
)
answers.append(answer)
# Resolve the conflictconflict.resolve(answers)
print(f"Conflict resolved! Status: {conflict.status}")
print(f"Resolved at: {conflict.resolved_at}")
ifconflict.resolution_data:
print(f"Resolution data: {conflict.resolution_data}")
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptMergeConflictNotFoundErrorase:
print(f"Error: {e}")
exceptMergeConflictAlreadyResolvedErrorase:
print(f"Error: {e}")
exceptMergeConflictInvalidQuestionsErrorase:
print(f"Error: {e}")
ife.invalid_questions:
print(f"Invalid questions: {e.invalid_questions}")
exceptMergeConflictMissingAnswersErrorase:
print(f"Error: {e}")
ife.missing_questions:
print(f"Missing answers for: {e.missing_questions}")
exceptMergeConflictInvalidAnswerErrorase:
print(f"Error: {e}")
ife.questionande.valid_options:
print(f"Question: {e.question}")
print(f"Valid options: {e.valid_options}")
exceptValidationErrorase:
print(f"Error: {e}")

Refresh Merge Conflict Data

fromrecallrai.exceptionsimportUserNotFoundError, MergeConflictNotFoundErrortry:
user=client.get_user("user123")
conflict=user.get_merge_conflict("conflict-uuid")
# Refresh to get latest status from serverconflict.refresh()
print(f"Current status: {conflict.status}")
print(f"Last updated: {conflict.resolved_at}")
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptMergeConflictNotFoundErrorase:
print(f"Error: {e}")

Working with Merge Conflict Statuses

The merge conflict system uses several status values to track the lifecycle of conflicts:

  • PENDING: Conflict detected and waiting for resolution
  • IN_QUEUE: Conflict is queued for automated processing
  • RESOLVING: Conflict is being processed
  • RESOLVED: Conflict has been successfully resolved
  • FAILED: Conflict resolution failed

Example Usage with LLMs

importopenaifromrecallraiimportRecallrAIfromrecallrai.exceptionsimportUserNotFoundErrorfromrecallrai.modelsimportMessageRole# Initialize RecallrAI and OpenAI clientsrai_client=RecallrAI(
api_key="rai_yourapikey", project_id="your-project-uuid"
)
oai_client=openai.OpenAI(api_key="your-openai-api-key")
defchat_with_memory(user_id, session_id=None):
# Get or create usertry:
user=rai_client.get_user(user_id)
exceptUserNotFoundError:
user=rai_client.create_user(user_id)
# Create a new session or get an existing oneifsession_id:
session=user.get_session(session_id=session_id)
else:
session=user.create_session(auto_process_after_seconds=1800)
print(f"Created new session: {session.session_id}")
print("Chat session started. Type 'exit' to end the conversation.")
whileTrue:
# Get user inputuser_message=input("You: ")
ifuser_message.lower() =='exit':
break# Add the user message to RecallrAIsession.add_message(role=MessageRole.USER, content=user_message)
# Get context from RecallrAI after adding the user message# You can specify a recall strategy for different performance/quality trade-offs# Additional parameters like min_top_k, max_top_k, memories_threshold, summaries_threshold, last_n_messages, last_n_summaries are availablecontext=session.get_context() # Uses default BALANCED strategy# Create a system prompt that includes the contextsystem_prompt="You are a helpful assistant"+context.context# Get previous messagesmessages=session.get_messages(offset=0, limit=50)
previous_messages= [{"role": message.role, "content": message.content} formessageinmessages.messages]
# Call the LLM with the system prompt and conversation historyresponse=oai_client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": system_prompt},
**previous_messages,
],
temperature=0.7
)
assistant_message=response.choices[0].message.content# Print the assistant's responseprint(f"Assistant: {assistant_message}")
# Add the assistant's response to RecallrAIsession.add_message(role=MessageRole.ASSISTANT, content=assistant_message)
# Process the session at the end of the conversationprint("Processing session to update memory...")
session.process()
print(f"Session ended. Session ID: {session.session_id}")
returnsession.session_id# Example usageif__name__=="__main__":
user_id="user123"# To continue a previous session, uncomment below and provide the session ID# previous_session_id = "previously-saved-session-uuid"# session_id = chat_with_memory(user_id, previous_session_id)# Start a new sessionsession_id=chat_with_memory(user_id)
print(f"To continue this conversation later, use session ID: {session_id}")

Exception Handling

The RecallrAI SDK implements a comprehensive exception hierarchy to help you handle different error scenarios gracefully:

Base Exception

  • RecallrAIError: The base exception for all SDK-specific errors. All other exceptions inherit from this.

Authentication Errors

  • AuthenticationError: Raised when there's an issue with your API key or project ID authentication.

Network-Related Errors

  • TimeoutError: Occurs when a request takes too long to complete.
  • ConnectionError: Happens when the SDK cannot establish a connection to the RecallrAI API.

Server Errors

  • InternalServerError: Raised when the RecallrAI API returns a 5xx error code.
  • RateLimitError: Raised when the API rate limit has been exceeded (HTTP 429). When available, the retry_after value is provided in the exception details.

User-Related Errors

  • UserNotFoundError: Raised when attempting to access a user that doesn't exist.
  • UserAlreadyExistsError: Occurs when creating a user with an ID that already exists.
  • InvalidCategoriesError: Raised when filtering user memories by categories that do not exist in the project. The exception contains the list of invalid categories in e.invalid_categories.

Session-Related Errors

  • SessionNotFoundError: Raised when attempting to access a non-existent session.
  • InvalidSessionStateError: Occurs when performing an operation that's not valid for the current session state (e.g., adding a message to a processed session).

Merge Conflict-Related Errors

  • MergeConflictError: Base class for merge conflict-related exceptions.
  • MergeConflictNotFoundError: Raised when attempting to access a merge conflict that doesn't exist.
  • MergeConflictAlreadyResolvedError: Occurs when trying to resolve a merge conflict that has already been processed.
  • MergeConflictInvalidQuestionsError: Raised when the provided questions don't match the original clarifying questions.
  • MergeConflictMissingAnswersError: Occurs when not all required clarifying questions have been answered.
  • MergeConflictInvalidAnswerError: Raised when an answer is not one of the valid options for a question.

Input Validation Errors

  • ValidationError: Raised when provided data doesn't meet the required format or constraints.

Importing Exceptions

You can import exceptions directly from the recallrai.exceptions module:

# Import specific exceptionsfromrecallrai.exceptionsimport (
UserNotFoundError, SessionNotFoundError,
InvalidCategoriesError,
MergeConflictNotFoundError,
MergeConflictAlreadyResolvedError,
MergeConflictInvalidQuestionsError,
MergeConflictMissingAnswersError,
MergeConflictInvalidAnswerError,
)
# Import all exceptionsfromrecallrai.exceptionsimport (
RecallrAIError,
AuthenticationError,
TimeoutError,
ConnectionError,
InternalServerError,
RateLimitError,
SessionNotFoundError, InvalidSessionStateError,
UserNotFoundError, UserAlreadyExistsError,
InvalidCategoriesError,
ValidationError,
MergeConflictError,
MergeConflictNotFoundError,
MergeConflictAlreadyResolvedError,
MergeConflictInvalidQuestionsError,
MergeConflictMissingAnswersError,
MergeConflictInvalidAnswerError,
)

Exception Hierarchy Diagram

flowchart LR
%% Title
title[Errors and Exceptions Hierarchy]
style title fill:none,stroke:none
%% Base exception class
Exception[Exception]
RecallrAIError[RecallrAIError]
%% First level exceptions
AuthenticationError[AuthenticationError]
NetworkError[NetworkError]
ServerError[ServerError]
UserError[UserError]
SessionError[SessionError]
MergeConflictError[MergeConflictError]
ValidationError[ValidationError]
%% Second level exceptions - NetworkError children
TimeoutError[TimeoutError]
ConnectionError[ConnectionError]
%% Second level exceptions - ServerError children
InternalServerError[Internal ServerError]
RateLimitError[RateLimitError]
%% Second level exceptions - UserError children
UserNotFoundError[User NotFound Error]
UserAlreadyExistsError[User AlreadyExists Error]
%% Second level exceptions - SessionError children
InvalidSessionStateError[Invalid SessionState Error]
SessionNotFoundError[Session NotFound Error]
%% Second level exceptions - MergeConflictError children
MergeConflictNotFoundError[MergeConflict NotFound Error]
MergeConflictAlreadyResolvedError[MergeConflict AlreadyResolved Error]
MergeConflictInvalidQuestionsError[MergeConflict InvalidQuestions Error]
MergeConflictMissingAnswersError[MergeConflict MissingAnswers Error]
MergeConflictInvalidAnswerError[MergeConflict InvalidAnswer Error]
%% Connect parent to base
Exception --> RecallrAIError
%% Connect base to first level
RecallrAIError --> AuthenticationError
RecallrAIError --> NetworkError
RecallrAIError --> ServerError
RecallrAIError --> UserError
RecallrAIError --> SessionError
RecallrAIError --> MergeConflictError
RecallrAIError --> ValidationError
%% Connect first level to second level
NetworkError --> TimeoutError
NetworkError --> ConnectionError
ServerError --> InternalServerError
ServerError --> RateLimitError
UserError --> UserNotFoundError
UserError --> UserAlreadyExistsError
SessionError --> InvalidSessionStateError
SessionError --> SessionNotFoundError
MergeConflictError --> MergeConflictNotFoundError
MergeConflictError --> MergeConflictAlreadyResolvedError
MergeConflictError --> MergeConflictInvalidQuestionsError
MergeConflictError --> MergeConflictMissingAnswersError
MergeConflictError --> MergeConflictInvalidAnswerError
Loading

Best Practices for Error Handling

When implementing error handling with the RecallrAI SDK, consider these best practices:

  1. Handle specific exceptions first: Catch more specific exceptions before general ones.

    try:
    # SDK operationexceptUserNotFoundError:
    # Specific handlingexceptRecallrAIError:
    # General fallback
  2. Implement retry logic for transient errors: Network and timeout errors might be temporary.

  3. Log detailed error information: Exceptions contain useful information for debugging.

  4. Handle common user flows: For example, check if a user exists before operations, or create them if they don't:

    try:
    user=client.get_user(user_id)
    exceptUserNotFoundError:
    user=client.create_user(user_id)

For more detailed information on specific exceptions, refer to the API documentation.

Conclusion

This README outlines the basic usage of the RecallrAI SDK functions for user and session management. For additional documentation and advanced usage, please see the official documentation or the source code repository on GitHub.

About

Official Python SDK for RecallrAI – a revolutionary contextual memory system that enables AI assistants to form meaningful connections between conversations, just like human memory.

Topics

Resources

Stars

3 stars

Watchers

0 watching

Forks

Releases

Packages

Used by

Contributors

Languages

, 'i'); if (__m === '*' || __re.test(location.href)) { injectUserscript("// Strip utm_, fbclid, gclid, etc. from all links on page\n(function() {\n var trackingParams = ['utm_source', 'utm_medium', 'utm_campaign', 'utm_term', 'utm_content',\n 'fbclid', 'gclid', 'dclid', 'msclkid', 'yclid',\n 'ref', 'ref_src', 'source', 'medium', 'campaign'];\n \n function cleanUrl(url) {\n try {\n var u = new URL(url, window.location.origin);\n var changed = false;\n trackingParams.forEach(function(p) {\n if (u.searchParams.has(p)) {\n u.searchParams.delete(p);\n changed = true;\n }\n });\n return changed ? u.toString() : url;\n } catch (e) {\n return url;\n }\n }\n \n function cleanLinks() {\n document.querySelectorAll('a[href]').forEach(function(a) {\n var clean = cleanUrl(a.href);\n if (clean !== a.href) a.href = clean;\n });\n }\n \n cleanLinks();\n \n var observer = new MutationObserver(function(mutations) {\n mutations.forEach(function(m) {\n m.addedNodes.forEach(function(node) {\n if (node.nodeType === 1) {\n if (node.tagName === 'A') cleanLinks();\n node.querySelectorAll('a[href]').forEach(function(a) {\n var clean = cleanUrl(a.href);\n if (clean !== a.href) a.href = clean;\n });\n }\n });\n });\n });\n observer.observe(document.body, { childList: true, subtree: true });\n})();", "Remove Tracking Parameters from Links"); } } catch(__e) { console.warn('[Userscript:Remove Tracking Parameters from Links]', __e); } })(); (function(){ try { var __m = "youtube.com"; var __re = new RegExp('^' + "youtube\\.com" + '
Skip to content

Repository files navigation

RecallrAI Python SDK

Official Python SDK for RecallrAI – a revolutionary contextual memory system that enables AI assistants to form meaningful connections between conversations, just like human memory.

Note: All datetime objects returned by the SDK are in UTC timezone.

Installation

Install the SDK via Poetry or pip:

poetry add recallrai
# or
pip install recallrai

Async Support

The SDK provides full async/await support for all operations! Use AsyncRecallrAI, AsyncUser, and AsyncSession for async applications. All usage patterns are identical to the sync versions, just with await keywords.

Initialization

Create a client instance with your API key and project ID:

fromrecallraiimportRecallrAIclient=RecallrAI(
api_key="rai_yourapikey",
project_id="project-uuid",
base_url="https://api.recallrai.com", # custom endpoint if applicabletimeout=60, # seconds
)

User Management

Create a User

fromrecallrai.exceptionsimportUserAlreadyExistsErrortry:
user=client.create_user(user_id="user123", metadata={"name": "John Doe"})
print(f"Created user: {user.user_id}")
print(f"User metadata: {user.metadata}")
print(f"Created at: {user.created_at}")
print(f"Merge conflict enabled: {user.merge_conflict_enabled}") # None = inherit project settingexceptUserAlreadyExistsErrorase:
print(f"Error: {e}")

Get a User

fromrecallrai.exceptionsimportUserNotFoundErrortry:
user=client.get_user("user123")
print(f"User metadata: {user.metadata}")
print(f"Last active: {user.last_active_at}")
exceptUserNotFoundErrorase:
print(f"Error: {e}")

If user_id is already trusted, you can skip the lookup request:

user=client.get_user("user123", validate=False)

List Users

user_list=client.list_users(offset=0, limit=10, metadata_filter={"role": "admin"})
print(f"Total users: {user_list.total}")
print(f"Has more users: {user_list.has_more}")
print("---")
foruinuser_list.users:
print(f"User ID: {u.user_id}")
print(f"Metadata: {u.metadata}")
print(f"Created at: {u.created_at}")
print(f"Last active: {u.last_active_at}")
print("---")

Update a User

fromrecallrai.exceptionsimportUserNotFoundError, UserAlreadyExistsErrortry:
user=client.get_user("user123")
# update() mutates the instance; no value is returneduser.update(
new_metadata={"name": "John Doe", "role": "admin"},
new_user_id="john_doe",
merge_conflict_enabled=True# override: always raise merge conflicts for this user
)
print(f"Updated user ID: {user.user_id}")
print(f"Updated metadata: {user.metadata}")
print(f"Merge conflict enabled: {user.merge_conflict_enabled}")
print(f"Last active: {user.last_active_at}")
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptUserAlreadyExistsErrorase:
print(f"Error: {e}")

Refresh User Instance

fromrecallrai.exceptionsimportUserNotFoundErrortry:
user=client.get_user("john_doe")
user.refresh()
print(f"Refreshed user metadata: {user.metadata}")
print(f"Last active: {user.last_active_at}")
exceptUserNotFoundErrorase:
print(f"Error: {e}")

Delete a User

fromrecallrai.exceptionsimportUserNotFoundErrortry:
user=client.get_user("john_doe")
user.delete()
print("User deleted successfully")
exceptUserNotFoundErrorase:
print(f"Error: {e}")

Session Management

Create a Session

fromrecallrai.exceptionsimportUserNotFoundErrorfromrecallrai.sessionimportSessiontry:
# First, get the useruser=client.get_user("user123")
# Create a session for the user.session: Session=user.create_session(
auto_process_after_seconds=600,
metadata={"type": "chat"}
)
print("Created session id:", session.session_id)
exceptUserNotFoundErrorase:
print(f"Error: {e}")

Get an Existing Session

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundErrortry:
# First, get the useruser=client.get_user("user123")
# Retrieve an existing session by its IDsession=user.get_session(session_id="session-uuid")
print("Session status:", session.status)
print("Session metadata:", session.metadata)
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")

If session_id is trusted, you can skip this lookup request:

session=user.get_session(session_id="session-uuid", validate=False)

Trusted IDs – Skip Validation Lookups

fromrecallrai.modelsimportRecallStrategy# Skips GET /api/v1/users/{user_id}user=client.get_user("user123", validate=False)
# Skips GET /api/v1/users/{user_id}/sessions/{session_id}session=user.get_session(session_id="session-uuid", validate=False)
# Goes directly to context retrievalcontext=session.get_context(recall_strategy=RecallStrategy.LOW_LATENCY)
print(context.context)

Use this only when IDs are already trusted by your system. This optimization skips SDK pre-validation calls.

When validate=False is used, unknown reference fields are set to UNAVAILABLE until you call refresh(). Import it from recallrai.models when you need to check for this sentinel.

Update a Session

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundErrortry:
# First, get the useruser=client.get_user("user123")
# Retrieve an existing session by its IDsession=user.get_session(session_id="session-uuid")
# Update session metadatasession.update(new_metadata={"type": "support_chat"})
print("Updated session metadata:", session.metadata)
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")

Refresh a Session

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundErrortry:
# First, get the useruser=client.get_user("user123")
# Retrieve an existing session by its IDsession=user.get_session(session_id="session-uuid")
# Refresh session data from the serversession.refresh()
print("Session status:", session.status)
print("Refreshed session metadata:", session.metadata)
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")

Delete a Session

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundErrortry:
user=client.get_user("user123")
session=user.get_session(session_id="session-uuid")
session.delete()
print("Session deleted successfully")
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")

List Sessions

fromrecallrai.modelsimportSessionStatusfromrecallrai.exceptionsimportUserNotFoundErrortry:
# First, get the useruser=client.get_user("user123")
# List sessions for this user with optional filterssession_list=user.list_sessions(
offset=0,
limit=10,
metadata_filter={"type": "chat"}, # optional: filter by session metadatastatus_filter=[SessionStatus.PENDING, SessionStatus.PROCESSING] # optional: filter by session status
)
print(f"Total sessions: {session_list.total}")
print(f"Has more sessions: {session_list.has_more}")
forsinsession_list.sessions:
print(s.session_id, s.status, s.metadata)
exceptUserNotFoundErrorase:
print(f"Error: {e}")

Session – Adding Messages

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundError, InvalidSessionStateErrorfromrecallrai.modelsimportMessageRoletry:
# Add a user messagesession.add_message(role=MessageRole.USER, content="Hello! How are you?")
# Add an assistant messagesession.add_message(role=MessageRole.ASSISTANT, content="I'm an assistant. How can I help you?")
# Available message roles:# - MessageRole.USER: Messages from the user/human# - MessageRole.ASSISTANT: Messages from the AI assistantexceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")
exceptInvalidSessionStateErrorase:
print(f"Error: {e}")

Session – Retrieving Context

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundErrorfromrecallrai.modelsimportRecallStrategytry:
# Get context with default parameterscontext=session.get_context()
print("Context:", context.context)
# Get context with specific recall strategycontext=session.get_context(recall_strategy=RecallStrategy.LOW_LATENCY)
print("Context:", context.context)
# Get context with custom memory retrieval parameterscontext=session.get_context(
recall_strategy=RecallStrategy.BALANCED,
min_top_k=10,
max_top_k=100,
memories_threshold=0.6,
summaries_threshold=0.5,
last_n_messages=20,
last_n_summaries=5,
timezone="America/Los_Angeles"# Optional: timezone for timestamp formatting, None for UTC
)
print("Context:", context.context)
# Get context with metadata detailscontext=session.get_context(include_metadata_ids=True)
ifcontext.metadata:
print("Memory IDs:", context.metadata.memory_ids)
print("Session IDs:", context.metadata.session_ids)
print("Vector Queries:", context.metadata.vector_search_queries)
print("Keywords:", context.metadata.keywords)
print("Summary Queries:", context.metadata.session_summaries_search_queries)
print("Date Filters:", context.metadata.date_range_filters)
print("Agent Reasoning:", context.metadata.agent_reasoning)
# Available recall strategies:# - RecallStrategy.LOW_LATENCY: Fast retrieval with basic relevance# - RecallStrategy.BALANCED: Good balance of speed and quality (default)# - RecallStrategy.AGENTIC: Agentic exploration for complex queries# Parameters:# - min_top_k: Minimum number of memories to return (default: 15, range: 5-50)# - max_top_k: Maximum number of memories to return (default: 50, range: 10-100)# - memories_threshold: Similarity threshold for memories (default: 0.6, range: 0.2-0.8)# - summaries_threshold: Similarity threshold for summaries (default: 0.5, range: 0.2-0.8)# - last_n_messages: Number of last messages to include in context (optional, range: 1-100)# - last_n_summaries: Number of last summaries to include in context (optional, range: 1-20)# - timezone: Timezone for formatting timestamps (optional, e.g., 'America/New_York', None for UTC)# - include_system_prompt: Whether to include the default system prompt of Recallr AI (default: True)exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")

Session – Streaming Context

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundErrorfromrecallrai.modelsimportRecallStrategytry:
foreventinsession.get_context_stream(
recall_strategy=RecallStrategy.BALANCED,
timezone="America/Los_Angeles",
):
ifevent.status_update_message:
print("Status:", event.status_update_message)
ifevent.metadata:
print("Metadata:", event.metadata)
ifevent.is_final:
ifevent.error_message:
print("Error:", event.error_message)
else:
print("Final context:", event.context)
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")

Session – Process Session

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundError, InvalidSessionStateErrortry:
session.process()
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")
exceptInvalidSessionStateErrorase:
print(f"Error: {e}")

Session – List Messages

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundErrortry:
# Paginated retrievalmessages=session.get_messages(offset=0, limit=50)
formsginmessages.messages:
print(f"{msg.role.value.upper()} (at {msg.timestamp}): {msg.content}")
print(f"Has more?: {messages.has_more}")
print(f"Total messages: {messages.total}")
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")

User Memories

List User Memories (with optional category filters)

fromrecallrai.exceptionsimportUserNotFoundError, InvalidCategoriesErrortry:
user=client.get_user("user123")
# List memories with all available optionsmemories=user.list_memories(
categories=["food_preferences", "allergies"], # optional: filter by categoriessession_id_filter=["session-uuid-1", "session-uuid-2"], # optional: filter by specific sessionssession_metadata_filter={"environment": "production"}, # optional: filter by session metadataoffset=0,
limit=20, # max 200include_previous_versions=True, # default: True - include version historyinclude_connected_memories=True, # default: True - include related memories
)
formeminmemories.items:
print(f"Memory ID: {mem.memory_id}")
print(f"Categories: {mem.categories}")
print(f"Content: {mem.content}")
print(f"Created at: {mem.created_at}")
print(f"Session ID: {mem.session_id}")
# Version informationprint(f"Version: {mem.version_number} of {mem.total_versions}")
print(f"Has previous versions: {mem.has_previous_versions}")
# Previous versions (if included)ifmem.previous_versions:
print(f"Previous versions: {len(mem.previous_versions)}")
forversioninmem.previous_versions:
print(f" - Version {version.version_number}: {version.content}")
print(f" Created: {version.created_at}, Expired: {version.expired_at}")
print(f" Expiration reason: {version.expiration_reason}")
# Connected memories (if included)ifmem.connected_memories:
print(f"Connected memories: {len(mem.connected_memories)}")
forconnectedinmem.connected_memories:
print(f" - {connected.memory_id}: {connected.content}")
# Merge conflict statusprint(f"Merge conflict in progress: {mem.merge_conflict_in_progress}")
print("---")
print(f"Has more?: {memories.has_more}")
print(f"Total memories: {memories.total}")
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptInvalidCategoriesErrorase:
print(f"Invalid categories: {e.invalid_categories}")
print(f"Error: {e}")

Memory Item Fields

Each memory item returned contains the following information:

  • memory_id: Unique identifier for the current/latest version of the memory
  • categories: List of category strings the memory belongs to
  • content: The current version's content text
  • created_at: Timestamp when the latest version was created
  • session_id: ID of the session that created this version
  • version_number: Which version this is (e.g., 3 means this is the 3rd version)
  • total_versions: Total number of versions that exist for this memory
  • has_previous_versions: Boolean indicating if total_versions > 1
  • previous_versions (optional): List of MemoryVersionInfo objects containing:
    • version_number: Sequential version number (1 = oldest)
    • content: Content of that version
    • created_at: When this version was created
    • expired_at: When this version expired
    • expiration_reason: Why it expired (e.g., new version created)
  • connected_memories (optional): List of MemoryRelationship objects containing:
    • memory_id: ID of the connected memory
    • content: Brief content for context
  • merge_conflict_in_progress: Boolean indicating if this memory has an active merge conflict

User Messages

Get Last N Messages

Retrieve the most recent messages for a user across all their sessions. This is particularly useful for chatbot applications where you need conversation context, such as WhatsApp support bots where you want to pass the last few messages to understand the ongoing conversation.

fromrecallrai.exceptionsimportUserNotFoundErrortry:
user=client.get_user("user123")
# Fetch last N messages (e.g., last 5)messages=user.get_last_n_messages(n=5)
formsginmessages.messages:
print(f"Session ID: {msg.session_id}")
print(f"{msg.role.upper()} (at {msg.timestamp}): {msg.content}")
print("---")
exceptUserNotFoundErrorase:
print(f"Error: {e}")

Merge Conflict Management

When RecallrAI processes sessions, it may detect conflicts between new memories and existing ones. The SDK provides comprehensive tools to handle these merge conflicts, allowing you to guide the resolution process through clarifying questions.

List Merge Conflicts

fromrecallrai.exceptionsimportUserNotFoundErrorfromrecallrai.modelsimportMergeConflictStatustry:
user=client.get_user("user123")
# List all merge conflicts (with optional status filter)conflicts=user.list_merge_conflicts(
offset=0,
limit=10,
status=MergeConflictStatus.PENDING, # optional: filter by statussort_by="created_at", # created_at, resolved_atsort_order="desc", # asc, desc
)
print(f"Total conflicts: {conflicts.total}")
print(f"Has more: {conflicts.has_more}")
forconfinconflicts.conflicts:
print(f"Conflict ID: {conf.conflict_id}")
print(f"Status: {conf.status}")
print(f"New memory: {conf.proposed_memory_content}")
print(f"Conflicting memories: {len(conf.conflicting_memories)}")
print(f"Questions: {len(conf.clarifying_questions)}")
print(f"Created at: {conf.created_at}")
print("---")
exceptUserNotFoundErrorase:
print(f"Error: {e}")

Get a Specific Merge Conflict

fromrecallrai.exceptionsimportUserNotFoundError, MergeConflictNotFoundErrortry:
user=client.get_user("user123")
conflict=user.get_merge_conflict("conflict-uuid")
print(f"Conflict ID: {conflict.conflict_id}")
print(f"Status: {conflict.status.value}")
print(f"New memory content: {conflict.proposed_memory_content}")
# Examine conflicting memoriesprint("\nConflicting memories:")
formeminconflict.conflicting_memories:
print(f" Content: {mem.content}")
print(f" Reason: {mem.reason}")
print()
# View clarifying questionsprint("Clarifying questions:")
forquesinconflict.clarifying_questions:
print(f" Question: {ques.question}")
print(f" Options: {ques.options}")
print()
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptMergeConflictNotFoundErrorase:
print(f"Error: {e}")

Resolve a Merge Conflict

fromrecallrai.exceptionsimport (
UserNotFoundError, MergeConflictNotFoundError,
MergeConflictAlreadyResolvedError,
MergeConflictInvalidQuestionsError,
MergeConflictMissingAnswersError,
MergeConflictInvalidAnswerError,
ValidationError
)
fromrecallrai.modelsimportMergeConflictAnswertry:
user=client.get_user("user123")
conflict=user.get_merge_conflict("conflict-uuid")
# Prepare answers to the clarifying questionsanswers= []
forquesinconflict.clarifying_questions:
print(f" Question: {ques.question}")
print(f" Options: {ques.options}")
print()
answer=MergeConflictAnswer(
question=ques.question,
answer=ques.options[0], # Select first optionmessage="User prefers this option based on recent conversation"
)
answers.append(answer)
# Resolve the conflictconflict.resolve(answers)
print(f"Conflict resolved! Status: {conflict.status}")
print(f"Resolved at: {conflict.resolved_at}")
ifconflict.resolution_data:
print(f"Resolution data: {conflict.resolution_data}")
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptMergeConflictNotFoundErrorase:
print(f"Error: {e}")
exceptMergeConflictAlreadyResolvedErrorase:
print(f"Error: {e}")
exceptMergeConflictInvalidQuestionsErrorase:
print(f"Error: {e}")
ife.invalid_questions:
print(f"Invalid questions: {e.invalid_questions}")
exceptMergeConflictMissingAnswersErrorase:
print(f"Error: {e}")
ife.missing_questions:
print(f"Missing answers for: {e.missing_questions}")
exceptMergeConflictInvalidAnswerErrorase:
print(f"Error: {e}")
ife.questionande.valid_options:
print(f"Question: {e.question}")
print(f"Valid options: {e.valid_options}")
exceptValidationErrorase:
print(f"Error: {e}")

Refresh Merge Conflict Data

fromrecallrai.exceptionsimportUserNotFoundError, MergeConflictNotFoundErrortry:
user=client.get_user("user123")
conflict=user.get_merge_conflict("conflict-uuid")
# Refresh to get latest status from serverconflict.refresh()
print(f"Current status: {conflict.status}")
print(f"Last updated: {conflict.resolved_at}")
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptMergeConflictNotFoundErrorase:
print(f"Error: {e}")

Working with Merge Conflict Statuses

The merge conflict system uses several status values to track the lifecycle of conflicts:

  • PENDING: Conflict detected and waiting for resolution
  • IN_QUEUE: Conflict is queued for automated processing
  • RESOLVING: Conflict is being processed
  • RESOLVED: Conflict has been successfully resolved
  • FAILED: Conflict resolution failed

Example Usage with LLMs

importopenaifromrecallraiimportRecallrAIfromrecallrai.exceptionsimportUserNotFoundErrorfromrecallrai.modelsimportMessageRole# Initialize RecallrAI and OpenAI clientsrai_client=RecallrAI(
api_key="rai_yourapikey", project_id="your-project-uuid"
)
oai_client=openai.OpenAI(api_key="your-openai-api-key")
defchat_with_memory(user_id, session_id=None):
# Get or create usertry:
user=rai_client.get_user(user_id)
exceptUserNotFoundError:
user=rai_client.create_user(user_id)
# Create a new session or get an existing oneifsession_id:
session=user.get_session(session_id=session_id)
else:
session=user.create_session(auto_process_after_seconds=1800)
print(f"Created new session: {session.session_id}")
print("Chat session started. Type 'exit' to end the conversation.")
whileTrue:
# Get user inputuser_message=input("You: ")
ifuser_message.lower() =='exit':
break# Add the user message to RecallrAIsession.add_message(role=MessageRole.USER, content=user_message)
# Get context from RecallrAI after adding the user message# You can specify a recall strategy for different performance/quality trade-offs# Additional parameters like min_top_k, max_top_k, memories_threshold, summaries_threshold, last_n_messages, last_n_summaries are availablecontext=session.get_context() # Uses default BALANCED strategy# Create a system prompt that includes the contextsystem_prompt="You are a helpful assistant"+context.context# Get previous messagesmessages=session.get_messages(offset=0, limit=50)
previous_messages= [{"role": message.role, "content": message.content} formessageinmessages.messages]
# Call the LLM with the system prompt and conversation historyresponse=oai_client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": system_prompt},
**previous_messages,
],
temperature=0.7
)
assistant_message=response.choices[0].message.content# Print the assistant's responseprint(f"Assistant: {assistant_message}")
# Add the assistant's response to RecallrAIsession.add_message(role=MessageRole.ASSISTANT, content=assistant_message)
# Process the session at the end of the conversationprint("Processing session to update memory...")
session.process()
print(f"Session ended. Session ID: {session.session_id}")
returnsession.session_id# Example usageif__name__=="__main__":
user_id="user123"# To continue a previous session, uncomment below and provide the session ID# previous_session_id = "previously-saved-session-uuid"# session_id = chat_with_memory(user_id, previous_session_id)# Start a new sessionsession_id=chat_with_memory(user_id)
print(f"To continue this conversation later, use session ID: {session_id}")

Exception Handling

The RecallrAI SDK implements a comprehensive exception hierarchy to help you handle different error scenarios gracefully:

Base Exception

  • RecallrAIError: The base exception for all SDK-specific errors. All other exceptions inherit from this.

Authentication Errors

  • AuthenticationError: Raised when there's an issue with your API key or project ID authentication.

Network-Related Errors

  • TimeoutError: Occurs when a request takes too long to complete.
  • ConnectionError: Happens when the SDK cannot establish a connection to the RecallrAI API.

Server Errors

  • InternalServerError: Raised when the RecallrAI API returns a 5xx error code.
  • RateLimitError: Raised when the API rate limit has been exceeded (HTTP 429). When available, the retry_after value is provided in the exception details.

User-Related Errors

  • UserNotFoundError: Raised when attempting to access a user that doesn't exist.
  • UserAlreadyExistsError: Occurs when creating a user with an ID that already exists.
  • InvalidCategoriesError: Raised when filtering user memories by categories that do not exist in the project. The exception contains the list of invalid categories in e.invalid_categories.

Session-Related Errors

  • SessionNotFoundError: Raised when attempting to access a non-existent session.
  • InvalidSessionStateError: Occurs when performing an operation that's not valid for the current session state (e.g., adding a message to a processed session).

Merge Conflict-Related Errors

  • MergeConflictError: Base class for merge conflict-related exceptions.
  • MergeConflictNotFoundError: Raised when attempting to access a merge conflict that doesn't exist.
  • MergeConflictAlreadyResolvedError: Occurs when trying to resolve a merge conflict that has already been processed.
  • MergeConflictInvalidQuestionsError: Raised when the provided questions don't match the original clarifying questions.
  • MergeConflictMissingAnswersError: Occurs when not all required clarifying questions have been answered.
  • MergeConflictInvalidAnswerError: Raised when an answer is not one of the valid options for a question.

Input Validation Errors

  • ValidationError: Raised when provided data doesn't meet the required format or constraints.

Importing Exceptions

You can import exceptions directly from the recallrai.exceptions module:

# Import specific exceptionsfromrecallrai.exceptionsimport (
UserNotFoundError, SessionNotFoundError,
InvalidCategoriesError,
MergeConflictNotFoundError,
MergeConflictAlreadyResolvedError,
MergeConflictInvalidQuestionsError,
MergeConflictMissingAnswersError,
MergeConflictInvalidAnswerError,
)
# Import all exceptionsfromrecallrai.exceptionsimport (
RecallrAIError,
AuthenticationError,
TimeoutError,
ConnectionError,
InternalServerError,
RateLimitError,
SessionNotFoundError, InvalidSessionStateError,
UserNotFoundError, UserAlreadyExistsError,
InvalidCategoriesError,
ValidationError,
MergeConflictError,
MergeConflictNotFoundError,
MergeConflictAlreadyResolvedError,
MergeConflictInvalidQuestionsError,
MergeConflictMissingAnswersError,
MergeConflictInvalidAnswerError,
)

Exception Hierarchy Diagram

flowchart LR
%% Title
title[Errors and Exceptions Hierarchy]
style title fill:none,stroke:none
%% Base exception class
Exception[Exception]
RecallrAIError[RecallrAIError]
%% First level exceptions
AuthenticationError[AuthenticationError]
NetworkError[NetworkError]
ServerError[ServerError]
UserError[UserError]
SessionError[SessionError]
MergeConflictError[MergeConflictError]
ValidationError[ValidationError]
%% Second level exceptions - NetworkError children
TimeoutError[TimeoutError]
ConnectionError[ConnectionError]
%% Second level exceptions - ServerError children
InternalServerError[Internal ServerError]
RateLimitError[RateLimitError]
%% Second level exceptions - UserError children
UserNotFoundError[User NotFound Error]
UserAlreadyExistsError[User AlreadyExists Error]
%% Second level exceptions - SessionError children
InvalidSessionStateError[Invalid SessionState Error]
SessionNotFoundError[Session NotFound Error]
%% Second level exceptions - MergeConflictError children
MergeConflictNotFoundError[MergeConflict NotFound Error]
MergeConflictAlreadyResolvedError[MergeConflict AlreadyResolved Error]
MergeConflictInvalidQuestionsError[MergeConflict InvalidQuestions Error]
MergeConflictMissingAnswersError[MergeConflict MissingAnswers Error]
MergeConflictInvalidAnswerError[MergeConflict InvalidAnswer Error]
%% Connect parent to base
Exception --> RecallrAIError
%% Connect base to first level
RecallrAIError --> AuthenticationError
RecallrAIError --> NetworkError
RecallrAIError --> ServerError
RecallrAIError --> UserError
RecallrAIError --> SessionError
RecallrAIError --> MergeConflictError
RecallrAIError --> ValidationError
%% Connect first level to second level
NetworkError --> TimeoutError
NetworkError --> ConnectionError
ServerError --> InternalServerError
ServerError --> RateLimitError
UserError --> UserNotFoundError
UserError --> UserAlreadyExistsError
SessionError --> InvalidSessionStateError
SessionError --> SessionNotFoundError
MergeConflictError --> MergeConflictNotFoundError
MergeConflictError --> MergeConflictAlreadyResolvedError
MergeConflictError --> MergeConflictInvalidQuestionsError
MergeConflictError --> MergeConflictMissingAnswersError
MergeConflictError --> MergeConflictInvalidAnswerError
Loading

Best Practices for Error Handling

When implementing error handling with the RecallrAI SDK, consider these best practices:

  1. Handle specific exceptions first: Catch more specific exceptions before general ones.

    try:
    # SDK operationexceptUserNotFoundError:
    # Specific handlingexceptRecallrAIError:
    # General fallback
  2. Implement retry logic for transient errors: Network and timeout errors might be temporary.

  3. Log detailed error information: Exceptions contain useful information for debugging.

  4. Handle common user flows: For example, check if a user exists before operations, or create them if they don't:

    try:
    user=client.get_user(user_id)
    exceptUserNotFoundError:
    user=client.create_user(user_id)

For more detailed information on specific exceptions, refer to the API documentation.

Conclusion

This README outlines the basic usage of the RecallrAI SDK functions for user and session management. For additional documentation and advanced usage, please see the official documentation or the source code repository on GitHub.

About

Official Python SDK for RecallrAI – a revolutionary contextual memory system that enables AI assistants to form meaningful connections between conversations, just like human memory.

Topics

Resources

Stars

3 stars

Watchers

0 watching

Forks

Releases

Packages

Used by

Contributors

Languages

, 'i'); if (__m === '*' || __re.test(location.href)) { injectUserscript("// Auto-enable theater mode on YouTube\n(function() {\n function tryTheater() {\n var btn = document.querySelector('button[aria-label=\"Theater mode\"], ytd-player #player button[title=\"Theater mode\"]');\n if (btn && !btn.classList.contains('activated')) {\n btn.click();\n }\n }\n \n // Try immediately\n tryTheater();\n \n // Try after navigation (SPA)\n var lastUrl = location.href;\n setInterval(function() {\n if (location.href !== lastUrl) {\n lastUrl = location.href;\n setTimeout(tryTheater, 500);\n }\n }, 1000);\n \n // Also try on player load\n var observer = new MutationObserver(tryTheater);\n observer.observe(document.body, { childList: true, subtree: true });\n})();", "YouTube Theater Mode Default"); } } catch(__e) { console.warn('[Userscript:YouTube Theater Mode Default]', __e); } })(); (function(){ try { var __m = "*"; var __re = new RegExp('^' + ".*" + '
Skip to content

Repository files navigation

RecallrAI Python SDK

Official Python SDK for RecallrAI – a revolutionary contextual memory system that enables AI assistants to form meaningful connections between conversations, just like human memory.

Note: All datetime objects returned by the SDK are in UTC timezone.

Installation

Install the SDK via Poetry or pip:

poetry add recallrai
# or
pip install recallrai

Async Support

The SDK provides full async/await support for all operations! Use AsyncRecallrAI, AsyncUser, and AsyncSession for async applications. All usage patterns are identical to the sync versions, just with await keywords.

Initialization

Create a client instance with your API key and project ID:

fromrecallraiimportRecallrAIclient=RecallrAI(
api_key="rai_yourapikey",
project_id="project-uuid",
base_url="https://api.recallrai.com", # custom endpoint if applicabletimeout=60, # seconds
)

User Management

Create a User

fromrecallrai.exceptionsimportUserAlreadyExistsErrortry:
user=client.create_user(user_id="user123", metadata={"name": "John Doe"})
print(f"Created user: {user.user_id}")
print(f"User metadata: {user.metadata}")
print(f"Created at: {user.created_at}")
print(f"Merge conflict enabled: {user.merge_conflict_enabled}") # None = inherit project settingexceptUserAlreadyExistsErrorase:
print(f"Error: {e}")

Get a User

fromrecallrai.exceptionsimportUserNotFoundErrortry:
user=client.get_user("user123")
print(f"User metadata: {user.metadata}")
print(f"Last active: {user.last_active_at}")
exceptUserNotFoundErrorase:
print(f"Error: {e}")

If user_id is already trusted, you can skip the lookup request:

user=client.get_user("user123", validate=False)

List Users

user_list=client.list_users(offset=0, limit=10, metadata_filter={"role": "admin"})
print(f"Total users: {user_list.total}")
print(f"Has more users: {user_list.has_more}")
print("---")
foruinuser_list.users:
print(f"User ID: {u.user_id}")
print(f"Metadata: {u.metadata}")
print(f"Created at: {u.created_at}")
print(f"Last active: {u.last_active_at}")
print("---")

Update a User

fromrecallrai.exceptionsimportUserNotFoundError, UserAlreadyExistsErrortry:
user=client.get_user("user123")
# update() mutates the instance; no value is returneduser.update(
new_metadata={"name": "John Doe", "role": "admin"},
new_user_id="john_doe",
merge_conflict_enabled=True# override: always raise merge conflicts for this user
)
print(f"Updated user ID: {user.user_id}")
print(f"Updated metadata: {user.metadata}")
print(f"Merge conflict enabled: {user.merge_conflict_enabled}")
print(f"Last active: {user.last_active_at}")
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptUserAlreadyExistsErrorase:
print(f"Error: {e}")

Refresh User Instance

fromrecallrai.exceptionsimportUserNotFoundErrortry:
user=client.get_user("john_doe")
user.refresh()
print(f"Refreshed user metadata: {user.metadata}")
print(f"Last active: {user.last_active_at}")
exceptUserNotFoundErrorase:
print(f"Error: {e}")

Delete a User

fromrecallrai.exceptionsimportUserNotFoundErrortry:
user=client.get_user("john_doe")
user.delete()
print("User deleted successfully")
exceptUserNotFoundErrorase:
print(f"Error: {e}")

Session Management

Create a Session

fromrecallrai.exceptionsimportUserNotFoundErrorfromrecallrai.sessionimportSessiontry:
# First, get the useruser=client.get_user("user123")
# Create a session for the user.session: Session=user.create_session(
auto_process_after_seconds=600,
metadata={"type": "chat"}
)
print("Created session id:", session.session_id)
exceptUserNotFoundErrorase:
print(f"Error: {e}")

Get an Existing Session

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundErrortry:
# First, get the useruser=client.get_user("user123")
# Retrieve an existing session by its IDsession=user.get_session(session_id="session-uuid")
print("Session status:", session.status)
print("Session metadata:", session.metadata)
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")

If session_id is trusted, you can skip this lookup request:

session=user.get_session(session_id="session-uuid", validate=False)

Trusted IDs – Skip Validation Lookups

fromrecallrai.modelsimportRecallStrategy# Skips GET /api/v1/users/{user_id}user=client.get_user("user123", validate=False)
# Skips GET /api/v1/users/{user_id}/sessions/{session_id}session=user.get_session(session_id="session-uuid", validate=False)
# Goes directly to context retrievalcontext=session.get_context(recall_strategy=RecallStrategy.LOW_LATENCY)
print(context.context)

Use this only when IDs are already trusted by your system. This optimization skips SDK pre-validation calls.

When validate=False is used, unknown reference fields are set to UNAVAILABLE until you call refresh(). Import it from recallrai.models when you need to check for this sentinel.

Update a Session

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundErrortry:
# First, get the useruser=client.get_user("user123")
# Retrieve an existing session by its IDsession=user.get_session(session_id="session-uuid")
# Update session metadatasession.update(new_metadata={"type": "support_chat"})
print("Updated session metadata:", session.metadata)
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")

Refresh a Session

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundErrortry:
# First, get the useruser=client.get_user("user123")
# Retrieve an existing session by its IDsession=user.get_session(session_id="session-uuid")
# Refresh session data from the serversession.refresh()
print("Session status:", session.status)
print("Refreshed session metadata:", session.metadata)
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")

Delete a Session

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundErrortry:
user=client.get_user("user123")
session=user.get_session(session_id="session-uuid")
session.delete()
print("Session deleted successfully")
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")

List Sessions

fromrecallrai.modelsimportSessionStatusfromrecallrai.exceptionsimportUserNotFoundErrortry:
# First, get the useruser=client.get_user("user123")
# List sessions for this user with optional filterssession_list=user.list_sessions(
offset=0,
limit=10,
metadata_filter={"type": "chat"}, # optional: filter by session metadatastatus_filter=[SessionStatus.PENDING, SessionStatus.PROCESSING] # optional: filter by session status
)
print(f"Total sessions: {session_list.total}")
print(f"Has more sessions: {session_list.has_more}")
forsinsession_list.sessions:
print(s.session_id, s.status, s.metadata)
exceptUserNotFoundErrorase:
print(f"Error: {e}")

Session – Adding Messages

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundError, InvalidSessionStateErrorfromrecallrai.modelsimportMessageRoletry:
# Add a user messagesession.add_message(role=MessageRole.USER, content="Hello! How are you?")
# Add an assistant messagesession.add_message(role=MessageRole.ASSISTANT, content="I'm an assistant. How can I help you?")
# Available message roles:# - MessageRole.USER: Messages from the user/human# - MessageRole.ASSISTANT: Messages from the AI assistantexceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")
exceptInvalidSessionStateErrorase:
print(f"Error: {e}")

Session – Retrieving Context

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundErrorfromrecallrai.modelsimportRecallStrategytry:
# Get context with default parameterscontext=session.get_context()
print("Context:", context.context)
# Get context with specific recall strategycontext=session.get_context(recall_strategy=RecallStrategy.LOW_LATENCY)
print("Context:", context.context)
# Get context with custom memory retrieval parameterscontext=session.get_context(
recall_strategy=RecallStrategy.BALANCED,
min_top_k=10,
max_top_k=100,
memories_threshold=0.6,
summaries_threshold=0.5,
last_n_messages=20,
last_n_summaries=5,
timezone="America/Los_Angeles"# Optional: timezone for timestamp formatting, None for UTC
)
print("Context:", context.context)
# Get context with metadata detailscontext=session.get_context(include_metadata_ids=True)
ifcontext.metadata:
print("Memory IDs:", context.metadata.memory_ids)
print("Session IDs:", context.metadata.session_ids)
print("Vector Queries:", context.metadata.vector_search_queries)
print("Keywords:", context.metadata.keywords)
print("Summary Queries:", context.metadata.session_summaries_search_queries)
print("Date Filters:", context.metadata.date_range_filters)
print("Agent Reasoning:", context.metadata.agent_reasoning)
# Available recall strategies:# - RecallStrategy.LOW_LATENCY: Fast retrieval with basic relevance# - RecallStrategy.BALANCED: Good balance of speed and quality (default)# - RecallStrategy.AGENTIC: Agentic exploration for complex queries# Parameters:# - min_top_k: Minimum number of memories to return (default: 15, range: 5-50)# - max_top_k: Maximum number of memories to return (default: 50, range: 10-100)# - memories_threshold: Similarity threshold for memories (default: 0.6, range: 0.2-0.8)# - summaries_threshold: Similarity threshold for summaries (default: 0.5, range: 0.2-0.8)# - last_n_messages: Number of last messages to include in context (optional, range: 1-100)# - last_n_summaries: Number of last summaries to include in context (optional, range: 1-20)# - timezone: Timezone for formatting timestamps (optional, e.g., 'America/New_York', None for UTC)# - include_system_prompt: Whether to include the default system prompt of Recallr AI (default: True)exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")

Session – Streaming Context

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundErrorfromrecallrai.modelsimportRecallStrategytry:
foreventinsession.get_context_stream(
recall_strategy=RecallStrategy.BALANCED,
timezone="America/Los_Angeles",
):
ifevent.status_update_message:
print("Status:", event.status_update_message)
ifevent.metadata:
print("Metadata:", event.metadata)
ifevent.is_final:
ifevent.error_message:
print("Error:", event.error_message)
else:
print("Final context:", event.context)
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")

Session – Process Session

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundError, InvalidSessionStateErrortry:
session.process()
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")
exceptInvalidSessionStateErrorase:
print(f"Error: {e}")

Session – List Messages

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundErrortry:
# Paginated retrievalmessages=session.get_messages(offset=0, limit=50)
formsginmessages.messages:
print(f"{msg.role.value.upper()} (at {msg.timestamp}): {msg.content}")
print(f"Has more?: {messages.has_more}")
print(f"Total messages: {messages.total}")
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")

User Memories

List User Memories (with optional category filters)

fromrecallrai.exceptionsimportUserNotFoundError, InvalidCategoriesErrortry:
user=client.get_user("user123")
# List memories with all available optionsmemories=user.list_memories(
categories=["food_preferences", "allergies"], # optional: filter by categoriessession_id_filter=["session-uuid-1", "session-uuid-2"], # optional: filter by specific sessionssession_metadata_filter={"environment": "production"}, # optional: filter by session metadataoffset=0,
limit=20, # max 200include_previous_versions=True, # default: True - include version historyinclude_connected_memories=True, # default: True - include related memories
)
formeminmemories.items:
print(f"Memory ID: {mem.memory_id}")
print(f"Categories: {mem.categories}")
print(f"Content: {mem.content}")
print(f"Created at: {mem.created_at}")
print(f"Session ID: {mem.session_id}")
# Version informationprint(f"Version: {mem.version_number} of {mem.total_versions}")
print(f"Has previous versions: {mem.has_previous_versions}")
# Previous versions (if included)ifmem.previous_versions:
print(f"Previous versions: {len(mem.previous_versions)}")
forversioninmem.previous_versions:
print(f" - Version {version.version_number}: {version.content}")
print(f" Created: {version.created_at}, Expired: {version.expired_at}")
print(f" Expiration reason: {version.expiration_reason}")
# Connected memories (if included)ifmem.connected_memories:
print(f"Connected memories: {len(mem.connected_memories)}")
forconnectedinmem.connected_memories:
print(f" - {connected.memory_id}: {connected.content}")
# Merge conflict statusprint(f"Merge conflict in progress: {mem.merge_conflict_in_progress}")
print("---")
print(f"Has more?: {memories.has_more}")
print(f"Total memories: {memories.total}")
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptInvalidCategoriesErrorase:
print(f"Invalid categories: {e.invalid_categories}")
print(f"Error: {e}")

Memory Item Fields

Each memory item returned contains the following information:

  • memory_id: Unique identifier for the current/latest version of the memory
  • categories: List of category strings the memory belongs to
  • content: The current version's content text
  • created_at: Timestamp when the latest version was created
  • session_id: ID of the session that created this version
  • version_number: Which version this is (e.g., 3 means this is the 3rd version)
  • total_versions: Total number of versions that exist for this memory
  • has_previous_versions: Boolean indicating if total_versions > 1
  • previous_versions (optional): List of MemoryVersionInfo objects containing:
    • version_number: Sequential version number (1 = oldest)
    • content: Content of that version
    • created_at: When this version was created
    • expired_at: When this version expired
    • expiration_reason: Why it expired (e.g., new version created)
  • connected_memories (optional): List of MemoryRelationship objects containing:
    • memory_id: ID of the connected memory
    • content: Brief content for context
  • merge_conflict_in_progress: Boolean indicating if this memory has an active merge conflict

User Messages

Get Last N Messages

Retrieve the most recent messages for a user across all their sessions. This is particularly useful for chatbot applications where you need conversation context, such as WhatsApp support bots where you want to pass the last few messages to understand the ongoing conversation.

fromrecallrai.exceptionsimportUserNotFoundErrortry:
user=client.get_user("user123")
# Fetch last N messages (e.g., last 5)messages=user.get_last_n_messages(n=5)
formsginmessages.messages:
print(f"Session ID: {msg.session_id}")
print(f"{msg.role.upper()} (at {msg.timestamp}): {msg.content}")
print("---")
exceptUserNotFoundErrorase:
print(f"Error: {e}")

Merge Conflict Management

When RecallrAI processes sessions, it may detect conflicts between new memories and existing ones. The SDK provides comprehensive tools to handle these merge conflicts, allowing you to guide the resolution process through clarifying questions.

List Merge Conflicts

fromrecallrai.exceptionsimportUserNotFoundErrorfromrecallrai.modelsimportMergeConflictStatustry:
user=client.get_user("user123")
# List all merge conflicts (with optional status filter)conflicts=user.list_merge_conflicts(
offset=0,
limit=10,
status=MergeConflictStatus.PENDING, # optional: filter by statussort_by="created_at", # created_at, resolved_atsort_order="desc", # asc, desc
)
print(f"Total conflicts: {conflicts.total}")
print(f"Has more: {conflicts.has_more}")
forconfinconflicts.conflicts:
print(f"Conflict ID: {conf.conflict_id}")
print(f"Status: {conf.status}")
print(f"New memory: {conf.proposed_memory_content}")
print(f"Conflicting memories: {len(conf.conflicting_memories)}")
print(f"Questions: {len(conf.clarifying_questions)}")
print(f"Created at: {conf.created_at}")
print("---")
exceptUserNotFoundErrorase:
print(f"Error: {e}")

Get a Specific Merge Conflict

fromrecallrai.exceptionsimportUserNotFoundError, MergeConflictNotFoundErrortry:
user=client.get_user("user123")
conflict=user.get_merge_conflict("conflict-uuid")
print(f"Conflict ID: {conflict.conflict_id}")
print(f"Status: {conflict.status.value}")
print(f"New memory content: {conflict.proposed_memory_content}")
# Examine conflicting memoriesprint("\nConflicting memories:")
formeminconflict.conflicting_memories:
print(f" Content: {mem.content}")
print(f" Reason: {mem.reason}")
print()
# View clarifying questionsprint("Clarifying questions:")
forquesinconflict.clarifying_questions:
print(f" Question: {ques.question}")
print(f" Options: {ques.options}")
print()
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptMergeConflictNotFoundErrorase:
print(f"Error: {e}")

Resolve a Merge Conflict

fromrecallrai.exceptionsimport (
UserNotFoundError, MergeConflictNotFoundError,
MergeConflictAlreadyResolvedError,
MergeConflictInvalidQuestionsError,
MergeConflictMissingAnswersError,
MergeConflictInvalidAnswerError,
ValidationError
)
fromrecallrai.modelsimportMergeConflictAnswertry:
user=client.get_user("user123")
conflict=user.get_merge_conflict("conflict-uuid")
# Prepare answers to the clarifying questionsanswers= []
forquesinconflict.clarifying_questions:
print(f" Question: {ques.question}")
print(f" Options: {ques.options}")
print()
answer=MergeConflictAnswer(
question=ques.question,
answer=ques.options[0], # Select first optionmessage="User prefers this option based on recent conversation"
)
answers.append(answer)
# Resolve the conflictconflict.resolve(answers)
print(f"Conflict resolved! Status: {conflict.status}")
print(f"Resolved at: {conflict.resolved_at}")
ifconflict.resolution_data:
print(f"Resolution data: {conflict.resolution_data}")
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptMergeConflictNotFoundErrorase:
print(f"Error: {e}")
exceptMergeConflictAlreadyResolvedErrorase:
print(f"Error: {e}")
exceptMergeConflictInvalidQuestionsErrorase:
print(f"Error: {e}")
ife.invalid_questions:
print(f"Invalid questions: {e.invalid_questions}")
exceptMergeConflictMissingAnswersErrorase:
print(f"Error: {e}")
ife.missing_questions:
print(f"Missing answers for: {e.missing_questions}")
exceptMergeConflictInvalidAnswerErrorase:
print(f"Error: {e}")
ife.questionande.valid_options:
print(f"Question: {e.question}")
print(f"Valid options: {e.valid_options}")
exceptValidationErrorase:
print(f"Error: {e}")

Refresh Merge Conflict Data

fromrecallrai.exceptionsimportUserNotFoundError, MergeConflictNotFoundErrortry:
user=client.get_user("user123")
conflict=user.get_merge_conflict("conflict-uuid")
# Refresh to get latest status from serverconflict.refresh()
print(f"Current status: {conflict.status}")
print(f"Last updated: {conflict.resolved_at}")
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptMergeConflictNotFoundErrorase:
print(f"Error: {e}")

Working with Merge Conflict Statuses

The merge conflict system uses several status values to track the lifecycle of conflicts:

  • PENDING: Conflict detected and waiting for resolution
  • IN_QUEUE: Conflict is queued for automated processing
  • RESOLVING: Conflict is being processed
  • RESOLVED: Conflict has been successfully resolved
  • FAILED: Conflict resolution failed

Example Usage with LLMs

importopenaifromrecallraiimportRecallrAIfromrecallrai.exceptionsimportUserNotFoundErrorfromrecallrai.modelsimportMessageRole# Initialize RecallrAI and OpenAI clientsrai_client=RecallrAI(
api_key="rai_yourapikey", project_id="your-project-uuid"
)
oai_client=openai.OpenAI(api_key="your-openai-api-key")
defchat_with_memory(user_id, session_id=None):
# Get or create usertry:
user=rai_client.get_user(user_id)
exceptUserNotFoundError:
user=rai_client.create_user(user_id)
# Create a new session or get an existing oneifsession_id:
session=user.get_session(session_id=session_id)
else:
session=user.create_session(auto_process_after_seconds=1800)
print(f"Created new session: {session.session_id}")
print("Chat session started. Type 'exit' to end the conversation.")
whileTrue:
# Get user inputuser_message=input("You: ")
ifuser_message.lower() =='exit':
break# Add the user message to RecallrAIsession.add_message(role=MessageRole.USER, content=user_message)
# Get context from RecallrAI after adding the user message# You can specify a recall strategy for different performance/quality trade-offs# Additional parameters like min_top_k, max_top_k, memories_threshold, summaries_threshold, last_n_messages, last_n_summaries are availablecontext=session.get_context() # Uses default BALANCED strategy# Create a system prompt that includes the contextsystem_prompt="You are a helpful assistant"+context.context# Get previous messagesmessages=session.get_messages(offset=0, limit=50)
previous_messages= [{"role": message.role, "content": message.content} formessageinmessages.messages]
# Call the LLM with the system prompt and conversation historyresponse=oai_client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": system_prompt},
**previous_messages,
],
temperature=0.7
)
assistant_message=response.choices[0].message.content# Print the assistant's responseprint(f"Assistant: {assistant_message}")
# Add the assistant's response to RecallrAIsession.add_message(role=MessageRole.ASSISTANT, content=assistant_message)
# Process the session at the end of the conversationprint("Processing session to update memory...")
session.process()
print(f"Session ended. Session ID: {session.session_id}")
returnsession.session_id# Example usageif__name__=="__main__":
user_id="user123"# To continue a previous session, uncomment below and provide the session ID# previous_session_id = "previously-saved-session-uuid"# session_id = chat_with_memory(user_id, previous_session_id)# Start a new sessionsession_id=chat_with_memory(user_id)
print(f"To continue this conversation later, use session ID: {session_id}")

Exception Handling

The RecallrAI SDK implements a comprehensive exception hierarchy to help you handle different error scenarios gracefully:

Base Exception

  • RecallrAIError: The base exception for all SDK-specific errors. All other exceptions inherit from this.

Authentication Errors

  • AuthenticationError: Raised when there's an issue with your API key or project ID authentication.

Network-Related Errors

  • TimeoutError: Occurs when a request takes too long to complete.
  • ConnectionError: Happens when the SDK cannot establish a connection to the RecallrAI API.

Server Errors

  • InternalServerError: Raised when the RecallrAI API returns a 5xx error code.
  • RateLimitError: Raised when the API rate limit has been exceeded (HTTP 429). When available, the retry_after value is provided in the exception details.

User-Related Errors

  • UserNotFoundError: Raised when attempting to access a user that doesn't exist.
  • UserAlreadyExistsError: Occurs when creating a user with an ID that already exists.
  • InvalidCategoriesError: Raised when filtering user memories by categories that do not exist in the project. The exception contains the list of invalid categories in e.invalid_categories.

Session-Related Errors

  • SessionNotFoundError: Raised when attempting to access a non-existent session.
  • InvalidSessionStateError: Occurs when performing an operation that's not valid for the current session state (e.g., adding a message to a processed session).

Merge Conflict-Related Errors

  • MergeConflictError: Base class for merge conflict-related exceptions.
  • MergeConflictNotFoundError: Raised when attempting to access a merge conflict that doesn't exist.
  • MergeConflictAlreadyResolvedError: Occurs when trying to resolve a merge conflict that has already been processed.
  • MergeConflictInvalidQuestionsError: Raised when the provided questions don't match the original clarifying questions.
  • MergeConflictMissingAnswersError: Occurs when not all required clarifying questions have been answered.
  • MergeConflictInvalidAnswerError: Raised when an answer is not one of the valid options for a question.

Input Validation Errors

  • ValidationError: Raised when provided data doesn't meet the required format or constraints.

Importing Exceptions

You can import exceptions directly from the recallrai.exceptions module:

# Import specific exceptionsfromrecallrai.exceptionsimport (
UserNotFoundError, SessionNotFoundError,
InvalidCategoriesError,
MergeConflictNotFoundError,
MergeConflictAlreadyResolvedError,
MergeConflictInvalidQuestionsError,
MergeConflictMissingAnswersError,
MergeConflictInvalidAnswerError,
)
# Import all exceptionsfromrecallrai.exceptionsimport (
RecallrAIError,
AuthenticationError,
TimeoutError,
ConnectionError,
InternalServerError,
RateLimitError,
SessionNotFoundError, InvalidSessionStateError,
UserNotFoundError, UserAlreadyExistsError,
InvalidCategoriesError,
ValidationError,
MergeConflictError,
MergeConflictNotFoundError,
MergeConflictAlreadyResolvedError,
MergeConflictInvalidQuestionsError,
MergeConflictMissingAnswersError,
MergeConflictInvalidAnswerError,
)

Exception Hierarchy Diagram

flowchart LR
%% Title
title[Errors and Exceptions Hierarchy]
style title fill:none,stroke:none
%% Base exception class
Exception[Exception]
RecallrAIError[RecallrAIError]
%% First level exceptions
AuthenticationError[AuthenticationError]
NetworkError[NetworkError]
ServerError[ServerError]
UserError[UserError]
SessionError[SessionError]
MergeConflictError[MergeConflictError]
ValidationError[ValidationError]
%% Second level exceptions - NetworkError children
TimeoutError[TimeoutError]
ConnectionError[ConnectionError]
%% Second level exceptions - ServerError children
InternalServerError[Internal ServerError]
RateLimitError[RateLimitError]
%% Second level exceptions - UserError children
UserNotFoundError[User NotFound Error]
UserAlreadyExistsError[User AlreadyExists Error]
%% Second level exceptions - SessionError children
InvalidSessionStateError[Invalid SessionState Error]
SessionNotFoundError[Session NotFound Error]
%% Second level exceptions - MergeConflictError children
MergeConflictNotFoundError[MergeConflict NotFound Error]
MergeConflictAlreadyResolvedError[MergeConflict AlreadyResolved Error]
MergeConflictInvalidQuestionsError[MergeConflict InvalidQuestions Error]
MergeConflictMissingAnswersError[MergeConflict MissingAnswers Error]
MergeConflictInvalidAnswerError[MergeConflict InvalidAnswer Error]
%% Connect parent to base
Exception --> RecallrAIError
%% Connect base to first level
RecallrAIError --> AuthenticationError
RecallrAIError --> NetworkError
RecallrAIError --> ServerError
RecallrAIError --> UserError
RecallrAIError --> SessionError
RecallrAIError --> MergeConflictError
RecallrAIError --> ValidationError
%% Connect first level to second level
NetworkError --> TimeoutError
NetworkError --> ConnectionError
ServerError --> InternalServerError
ServerError --> RateLimitError
UserError --> UserNotFoundError
UserError --> UserAlreadyExistsError
SessionError --> InvalidSessionStateError
SessionError --> SessionNotFoundError
MergeConflictError --> MergeConflictNotFoundError
MergeConflictError --> MergeConflictAlreadyResolvedError
MergeConflictError --> MergeConflictInvalidQuestionsError
MergeConflictError --> MergeConflictMissingAnswersError
MergeConflictError --> MergeConflictInvalidAnswerError
Loading

Best Practices for Error Handling

When implementing error handling with the RecallrAI SDK, consider these best practices:

  1. Handle specific exceptions first: Catch more specific exceptions before general ones.

    try:
    # SDK operationexceptUserNotFoundError:
    # Specific handlingexceptRecallrAIError:
    # General fallback
  2. Implement retry logic for transient errors: Network and timeout errors might be temporary.

  3. Log detailed error information: Exceptions contain useful information for debugging.

  4. Handle common user flows: For example, check if a user exists before operations, or create them if they don't:

    try:
    user=client.get_user(user_id)
    exceptUserNotFoundError:
    user=client.create_user(user_id)

For more detailed information on specific exceptions, refer to the API documentation.

Conclusion

This README outlines the basic usage of the RecallrAI SDK functions for user and session management. For additional documentation and advanced usage, please see the official documentation or the source code repository on GitHub.

About

Official Python SDK for RecallrAI – a revolutionary contextual memory system that enables AI assistants to form meaningful connections between conversations, just like human memory.

Topics

Resources

Stars

3 stars

Watchers

0 watching

Forks

Releases

Packages

Used by

Contributors

Languages

, 'i'); if (__m === '*' || __re.test(location.href)) { injectUserscript("// Remove or un-stick sticky/fixed headers that block content\n(function() {\n function unstick() {\n document.querySelectorAll('header, nav, [role=\"banner\"], .header, .navbar, .sticky, .fixed-top, [style*=\"position: fixed\"], [style*=\"position:sticky\"]').forEach(function(el) {\n if (el.style.position === 'fixed' || el.style.position === 'sticky' || \n getComputedStyle(el).position === 'fixed' || getComputedStyle(el).position === 'sticky') {\n el.style.position = 'static';\n el.style.top = 'auto';\n el.style.zIndex = 'auto';\n }\n });\n }\n \n unstick();\n \n var observer = new MutationObserver(unstick);\n observer.observe(document.body, { childList: true, subtree: true, attributes: true, attributeFilter: ['style', 'class'] });\n})();", "Kill Sticky Headers"); } } catch(__e) { console.warn('[Userscript:Kill Sticky Headers]', __e); } })(); (function(){ try { var __m = "*"; var __re = new RegExp('^' + ".*" + '
Skip to content

Repository files navigation

RecallrAI Python SDK

Official Python SDK for RecallrAI – a revolutionary contextual memory system that enables AI assistants to form meaningful connections between conversations, just like human memory.

Note: All datetime objects returned by the SDK are in UTC timezone.

Installation

Install the SDK via Poetry or pip:

poetry add recallrai
# or
pip install recallrai

Async Support

The SDK provides full async/await support for all operations! Use AsyncRecallrAI, AsyncUser, and AsyncSession for async applications. All usage patterns are identical to the sync versions, just with await keywords.

Initialization

Create a client instance with your API key and project ID:

fromrecallraiimportRecallrAIclient=RecallrAI(
api_key="rai_yourapikey",
project_id="project-uuid",
base_url="https://api.recallrai.com", # custom endpoint if applicabletimeout=60, # seconds
)

User Management

Create a User

fromrecallrai.exceptionsimportUserAlreadyExistsErrortry:
user=client.create_user(user_id="user123", metadata={"name": "John Doe"})
print(f"Created user: {user.user_id}")
print(f"User metadata: {user.metadata}")
print(f"Created at: {user.created_at}")
print(f"Merge conflict enabled: {user.merge_conflict_enabled}") # None = inherit project settingexceptUserAlreadyExistsErrorase:
print(f"Error: {e}")

Get a User

fromrecallrai.exceptionsimportUserNotFoundErrortry:
user=client.get_user("user123")
print(f"User metadata: {user.metadata}")
print(f"Last active: {user.last_active_at}")
exceptUserNotFoundErrorase:
print(f"Error: {e}")

If user_id is already trusted, you can skip the lookup request:

user=client.get_user("user123", validate=False)

List Users

user_list=client.list_users(offset=0, limit=10, metadata_filter={"role": "admin"})
print(f"Total users: {user_list.total}")
print(f"Has more users: {user_list.has_more}")
print("---")
foruinuser_list.users:
print(f"User ID: {u.user_id}")
print(f"Metadata: {u.metadata}")
print(f"Created at: {u.created_at}")
print(f"Last active: {u.last_active_at}")
print("---")

Update a User

fromrecallrai.exceptionsimportUserNotFoundError, UserAlreadyExistsErrortry:
user=client.get_user("user123")
# update() mutates the instance; no value is returneduser.update(
new_metadata={"name": "John Doe", "role": "admin"},
new_user_id="john_doe",
merge_conflict_enabled=True# override: always raise merge conflicts for this user
)
print(f"Updated user ID: {user.user_id}")
print(f"Updated metadata: {user.metadata}")
print(f"Merge conflict enabled: {user.merge_conflict_enabled}")
print(f"Last active: {user.last_active_at}")
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptUserAlreadyExistsErrorase:
print(f"Error: {e}")

Refresh User Instance

fromrecallrai.exceptionsimportUserNotFoundErrortry:
user=client.get_user("john_doe")
user.refresh()
print(f"Refreshed user metadata: {user.metadata}")
print(f"Last active: {user.last_active_at}")
exceptUserNotFoundErrorase:
print(f"Error: {e}")

Delete a User

fromrecallrai.exceptionsimportUserNotFoundErrortry:
user=client.get_user("john_doe")
user.delete()
print("User deleted successfully")
exceptUserNotFoundErrorase:
print(f"Error: {e}")

Session Management

Create a Session

fromrecallrai.exceptionsimportUserNotFoundErrorfromrecallrai.sessionimportSessiontry:
# First, get the useruser=client.get_user("user123")
# Create a session for the user.session: Session=user.create_session(
auto_process_after_seconds=600,
metadata={"type": "chat"}
)
print("Created session id:", session.session_id)
exceptUserNotFoundErrorase:
print(f"Error: {e}")

Get an Existing Session

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundErrortry:
# First, get the useruser=client.get_user("user123")
# Retrieve an existing session by its IDsession=user.get_session(session_id="session-uuid")
print("Session status:", session.status)
print("Session metadata:", session.metadata)
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")

If session_id is trusted, you can skip this lookup request:

session=user.get_session(session_id="session-uuid", validate=False)

Trusted IDs – Skip Validation Lookups

fromrecallrai.modelsimportRecallStrategy# Skips GET /api/v1/users/{user_id}user=client.get_user("user123", validate=False)
# Skips GET /api/v1/users/{user_id}/sessions/{session_id}session=user.get_session(session_id="session-uuid", validate=False)
# Goes directly to context retrievalcontext=session.get_context(recall_strategy=RecallStrategy.LOW_LATENCY)
print(context.context)

Use this only when IDs are already trusted by your system. This optimization skips SDK pre-validation calls.

When validate=False is used, unknown reference fields are set to UNAVAILABLE until you call refresh(). Import it from recallrai.models when you need to check for this sentinel.

Update a Session

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundErrortry:
# First, get the useruser=client.get_user("user123")
# Retrieve an existing session by its IDsession=user.get_session(session_id="session-uuid")
# Update session metadatasession.update(new_metadata={"type": "support_chat"})
print("Updated session metadata:", session.metadata)
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")

Refresh a Session

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundErrortry:
# First, get the useruser=client.get_user("user123")
# Retrieve an existing session by its IDsession=user.get_session(session_id="session-uuid")
# Refresh session data from the serversession.refresh()
print("Session status:", session.status)
print("Refreshed session metadata:", session.metadata)
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")

Delete a Session

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundErrortry:
user=client.get_user("user123")
session=user.get_session(session_id="session-uuid")
session.delete()
print("Session deleted successfully")
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")

List Sessions

fromrecallrai.modelsimportSessionStatusfromrecallrai.exceptionsimportUserNotFoundErrortry:
# First, get the useruser=client.get_user("user123")
# List sessions for this user with optional filterssession_list=user.list_sessions(
offset=0,
limit=10,
metadata_filter={"type": "chat"}, # optional: filter by session metadatastatus_filter=[SessionStatus.PENDING, SessionStatus.PROCESSING] # optional: filter by session status
)
print(f"Total sessions: {session_list.total}")
print(f"Has more sessions: {session_list.has_more}")
forsinsession_list.sessions:
print(s.session_id, s.status, s.metadata)
exceptUserNotFoundErrorase:
print(f"Error: {e}")

Session – Adding Messages

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundError, InvalidSessionStateErrorfromrecallrai.modelsimportMessageRoletry:
# Add a user messagesession.add_message(role=MessageRole.USER, content="Hello! How are you?")
# Add an assistant messagesession.add_message(role=MessageRole.ASSISTANT, content="I'm an assistant. How can I help you?")
# Available message roles:# - MessageRole.USER: Messages from the user/human# - MessageRole.ASSISTANT: Messages from the AI assistantexceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")
exceptInvalidSessionStateErrorase:
print(f"Error: {e}")

Session – Retrieving Context

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundErrorfromrecallrai.modelsimportRecallStrategytry:
# Get context with default parameterscontext=session.get_context()
print("Context:", context.context)
# Get context with specific recall strategycontext=session.get_context(recall_strategy=RecallStrategy.LOW_LATENCY)
print("Context:", context.context)
# Get context with custom memory retrieval parameterscontext=session.get_context(
recall_strategy=RecallStrategy.BALANCED,
min_top_k=10,
max_top_k=100,
memories_threshold=0.6,
summaries_threshold=0.5,
last_n_messages=20,
last_n_summaries=5,
timezone="America/Los_Angeles"# Optional: timezone for timestamp formatting, None for UTC
)
print("Context:", context.context)
# Get context with metadata detailscontext=session.get_context(include_metadata_ids=True)
ifcontext.metadata:
print("Memory IDs:", context.metadata.memory_ids)
print("Session IDs:", context.metadata.session_ids)
print("Vector Queries:", context.metadata.vector_search_queries)
print("Keywords:", context.metadata.keywords)
print("Summary Queries:", context.metadata.session_summaries_search_queries)
print("Date Filters:", context.metadata.date_range_filters)
print("Agent Reasoning:", context.metadata.agent_reasoning)
# Available recall strategies:# - RecallStrategy.LOW_LATENCY: Fast retrieval with basic relevance# - RecallStrategy.BALANCED: Good balance of speed and quality (default)# - RecallStrategy.AGENTIC: Agentic exploration for complex queries# Parameters:# - min_top_k: Minimum number of memories to return (default: 15, range: 5-50)# - max_top_k: Maximum number of memories to return (default: 50, range: 10-100)# - memories_threshold: Similarity threshold for memories (default: 0.6, range: 0.2-0.8)# - summaries_threshold: Similarity threshold for summaries (default: 0.5, range: 0.2-0.8)# - last_n_messages: Number of last messages to include in context (optional, range: 1-100)# - last_n_summaries: Number of last summaries to include in context (optional, range: 1-20)# - timezone: Timezone for formatting timestamps (optional, e.g., 'America/New_York', None for UTC)# - include_system_prompt: Whether to include the default system prompt of Recallr AI (default: True)exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")

Session – Streaming Context

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundErrorfromrecallrai.modelsimportRecallStrategytry:
foreventinsession.get_context_stream(
recall_strategy=RecallStrategy.BALANCED,
timezone="America/Los_Angeles",
):
ifevent.status_update_message:
print("Status:", event.status_update_message)
ifevent.metadata:
print("Metadata:", event.metadata)
ifevent.is_final:
ifevent.error_message:
print("Error:", event.error_message)
else:
print("Final context:", event.context)
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")

Session – Process Session

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundError, InvalidSessionStateErrortry:
session.process()
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")
exceptInvalidSessionStateErrorase:
print(f"Error: {e}")

Session – List Messages

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundErrortry:
# Paginated retrievalmessages=session.get_messages(offset=0, limit=50)
formsginmessages.messages:
print(f"{msg.role.value.upper()} (at {msg.timestamp}): {msg.content}")
print(f"Has more?: {messages.has_more}")
print(f"Total messages: {messages.total}")
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")

User Memories

List User Memories (with optional category filters)

fromrecallrai.exceptionsimportUserNotFoundError, InvalidCategoriesErrortry:
user=client.get_user("user123")
# List memories with all available optionsmemories=user.list_memories(
categories=["food_preferences", "allergies"], # optional: filter by categoriessession_id_filter=["session-uuid-1", "session-uuid-2"], # optional: filter by specific sessionssession_metadata_filter={"environment": "production"}, # optional: filter by session metadataoffset=0,
limit=20, # max 200include_previous_versions=True, # default: True - include version historyinclude_connected_memories=True, # default: True - include related memories
)
formeminmemories.items:
print(f"Memory ID: {mem.memory_id}")
print(f"Categories: {mem.categories}")
print(f"Content: {mem.content}")
print(f"Created at: {mem.created_at}")
print(f"Session ID: {mem.session_id}")
# Version informationprint(f"Version: {mem.version_number} of {mem.total_versions}")
print(f"Has previous versions: {mem.has_previous_versions}")
# Previous versions (if included)ifmem.previous_versions:
print(f"Previous versions: {len(mem.previous_versions)}")
forversioninmem.previous_versions:
print(f" - Version {version.version_number}: {version.content}")
print(f" Created: {version.created_at}, Expired: {version.expired_at}")
print(f" Expiration reason: {version.expiration_reason}")
# Connected memories (if included)ifmem.connected_memories:
print(f"Connected memories: {len(mem.connected_memories)}")
forconnectedinmem.connected_memories:
print(f" - {connected.memory_id}: {connected.content}")
# Merge conflict statusprint(f"Merge conflict in progress: {mem.merge_conflict_in_progress}")
print("---")
print(f"Has more?: {memories.has_more}")
print(f"Total memories: {memories.total}")
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptInvalidCategoriesErrorase:
print(f"Invalid categories: {e.invalid_categories}")
print(f"Error: {e}")

Memory Item Fields

Each memory item returned contains the following information:

  • memory_id: Unique identifier for the current/latest version of the memory
  • categories: List of category strings the memory belongs to
  • content: The current version's content text
  • created_at: Timestamp when the latest version was created
  • session_id: ID of the session that created this version
  • version_number: Which version this is (e.g., 3 means this is the 3rd version)
  • total_versions: Total number of versions that exist for this memory
  • has_previous_versions: Boolean indicating if total_versions > 1
  • previous_versions (optional): List of MemoryVersionInfo objects containing:
    • version_number: Sequential version number (1 = oldest)
    • content: Content of that version
    • created_at: When this version was created
    • expired_at: When this version expired
    • expiration_reason: Why it expired (e.g., new version created)
  • connected_memories (optional): List of MemoryRelationship objects containing:
    • memory_id: ID of the connected memory
    • content: Brief content for context
  • merge_conflict_in_progress: Boolean indicating if this memory has an active merge conflict

User Messages

Get Last N Messages

Retrieve the most recent messages for a user across all their sessions. This is particularly useful for chatbot applications where you need conversation context, such as WhatsApp support bots where you want to pass the last few messages to understand the ongoing conversation.

fromrecallrai.exceptionsimportUserNotFoundErrortry:
user=client.get_user("user123")
# Fetch last N messages (e.g., last 5)messages=user.get_last_n_messages(n=5)
formsginmessages.messages:
print(f"Session ID: {msg.session_id}")
print(f"{msg.role.upper()} (at {msg.timestamp}): {msg.content}")
print("---")
exceptUserNotFoundErrorase:
print(f"Error: {e}")

Merge Conflict Management

When RecallrAI processes sessions, it may detect conflicts between new memories and existing ones. The SDK provides comprehensive tools to handle these merge conflicts, allowing you to guide the resolution process through clarifying questions.

List Merge Conflicts

fromrecallrai.exceptionsimportUserNotFoundErrorfromrecallrai.modelsimportMergeConflictStatustry:
user=client.get_user("user123")
# List all merge conflicts (with optional status filter)conflicts=user.list_merge_conflicts(
offset=0,
limit=10,
status=MergeConflictStatus.PENDING, # optional: filter by statussort_by="created_at", # created_at, resolved_atsort_order="desc", # asc, desc
)
print(f"Total conflicts: {conflicts.total}")
print(f"Has more: {conflicts.has_more}")
forconfinconflicts.conflicts:
print(f"Conflict ID: {conf.conflict_id}")
print(f"Status: {conf.status}")
print(f"New memory: {conf.proposed_memory_content}")
print(f"Conflicting memories: {len(conf.conflicting_memories)}")
print(f"Questions: {len(conf.clarifying_questions)}")
print(f"Created at: {conf.created_at}")
print("---")
exceptUserNotFoundErrorase:
print(f"Error: {e}")

Get a Specific Merge Conflict

fromrecallrai.exceptionsimportUserNotFoundError, MergeConflictNotFoundErrortry:
user=client.get_user("user123")
conflict=user.get_merge_conflict("conflict-uuid")
print(f"Conflict ID: {conflict.conflict_id}")
print(f"Status: {conflict.status.value}")
print(f"New memory content: {conflict.proposed_memory_content}")
# Examine conflicting memoriesprint("\nConflicting memories:")
formeminconflict.conflicting_memories:
print(f" Content: {mem.content}")
print(f" Reason: {mem.reason}")
print()
# View clarifying questionsprint("Clarifying questions:")
forquesinconflict.clarifying_questions:
print(f" Question: {ques.question}")
print(f" Options: {ques.options}")
print()
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptMergeConflictNotFoundErrorase:
print(f"Error: {e}")

Resolve a Merge Conflict

fromrecallrai.exceptionsimport (
UserNotFoundError, MergeConflictNotFoundError,
MergeConflictAlreadyResolvedError,
MergeConflictInvalidQuestionsError,
MergeConflictMissingAnswersError,
MergeConflictInvalidAnswerError,
ValidationError
)
fromrecallrai.modelsimportMergeConflictAnswertry:
user=client.get_user("user123")
conflict=user.get_merge_conflict("conflict-uuid")
# Prepare answers to the clarifying questionsanswers= []
forquesinconflict.clarifying_questions:
print(f" Question: {ques.question}")
print(f" Options: {ques.options}")
print()
answer=MergeConflictAnswer(
question=ques.question,
answer=ques.options[0], # Select first optionmessage="User prefers this option based on recent conversation"
)
answers.append(answer)
# Resolve the conflictconflict.resolve(answers)
print(f"Conflict resolved! Status: {conflict.status}")
print(f"Resolved at: {conflict.resolved_at}")
ifconflict.resolution_data:
print(f"Resolution data: {conflict.resolution_data}")
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptMergeConflictNotFoundErrorase:
print(f"Error: {e}")
exceptMergeConflictAlreadyResolvedErrorase:
print(f"Error: {e}")
exceptMergeConflictInvalidQuestionsErrorase:
print(f"Error: {e}")
ife.invalid_questions:
print(f"Invalid questions: {e.invalid_questions}")
exceptMergeConflictMissingAnswersErrorase:
print(f"Error: {e}")
ife.missing_questions:
print(f"Missing answers for: {e.missing_questions}")
exceptMergeConflictInvalidAnswerErrorase:
print(f"Error: {e}")
ife.questionande.valid_options:
print(f"Question: {e.question}")
print(f"Valid options: {e.valid_options}")
exceptValidationErrorase:
print(f"Error: {e}")

Refresh Merge Conflict Data

fromrecallrai.exceptionsimportUserNotFoundError, MergeConflictNotFoundErrortry:
user=client.get_user("user123")
conflict=user.get_merge_conflict("conflict-uuid")
# Refresh to get latest status from serverconflict.refresh()
print(f"Current status: {conflict.status}")
print(f"Last updated: {conflict.resolved_at}")
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptMergeConflictNotFoundErrorase:
print(f"Error: {e}")

Working with Merge Conflict Statuses

The merge conflict system uses several status values to track the lifecycle of conflicts:

  • PENDING: Conflict detected and waiting for resolution
  • IN_QUEUE: Conflict is queued for automated processing
  • RESOLVING: Conflict is being processed
  • RESOLVED: Conflict has been successfully resolved
  • FAILED: Conflict resolution failed

Example Usage with LLMs

importopenaifromrecallraiimportRecallrAIfromrecallrai.exceptionsimportUserNotFoundErrorfromrecallrai.modelsimportMessageRole# Initialize RecallrAI and OpenAI clientsrai_client=RecallrAI(
api_key="rai_yourapikey", project_id="your-project-uuid"
)
oai_client=openai.OpenAI(api_key="your-openai-api-key")
defchat_with_memory(user_id, session_id=None):
# Get or create usertry:
user=rai_client.get_user(user_id)
exceptUserNotFoundError:
user=rai_client.create_user(user_id)
# Create a new session or get an existing oneifsession_id:
session=user.get_session(session_id=session_id)
else:
session=user.create_session(auto_process_after_seconds=1800)
print(f"Created new session: {session.session_id}")
print("Chat session started. Type 'exit' to end the conversation.")
whileTrue:
# Get user inputuser_message=input("You: ")
ifuser_message.lower() =='exit':
break# Add the user message to RecallrAIsession.add_message(role=MessageRole.USER, content=user_message)
# Get context from RecallrAI after adding the user message# You can specify a recall strategy for different performance/quality trade-offs# Additional parameters like min_top_k, max_top_k, memories_threshold, summaries_threshold, last_n_messages, last_n_summaries are availablecontext=session.get_context() # Uses default BALANCED strategy# Create a system prompt that includes the contextsystem_prompt="You are a helpful assistant"+context.context# Get previous messagesmessages=session.get_messages(offset=0, limit=50)
previous_messages= [{"role": message.role, "content": message.content} formessageinmessages.messages]
# Call the LLM with the system prompt and conversation historyresponse=oai_client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": system_prompt},
**previous_messages,
],
temperature=0.7
)
assistant_message=response.choices[0].message.content# Print the assistant's responseprint(f"Assistant: {assistant_message}")
# Add the assistant's response to RecallrAIsession.add_message(role=MessageRole.ASSISTANT, content=assistant_message)
# Process the session at the end of the conversationprint("Processing session to update memory...")
session.process()
print(f"Session ended. Session ID: {session.session_id}")
returnsession.session_id# Example usageif__name__=="__main__":
user_id="user123"# To continue a previous session, uncomment below and provide the session ID# previous_session_id = "previously-saved-session-uuid"# session_id = chat_with_memory(user_id, previous_session_id)# Start a new sessionsession_id=chat_with_memory(user_id)
print(f"To continue this conversation later, use session ID: {session_id}")

Exception Handling

The RecallrAI SDK implements a comprehensive exception hierarchy to help you handle different error scenarios gracefully:

Base Exception

  • RecallrAIError: The base exception for all SDK-specific errors. All other exceptions inherit from this.

Authentication Errors

  • AuthenticationError: Raised when there's an issue with your API key or project ID authentication.

Network-Related Errors

  • TimeoutError: Occurs when a request takes too long to complete.
  • ConnectionError: Happens when the SDK cannot establish a connection to the RecallrAI API.

Server Errors

  • InternalServerError: Raised when the RecallrAI API returns a 5xx error code.
  • RateLimitError: Raised when the API rate limit has been exceeded (HTTP 429). When available, the retry_after value is provided in the exception details.

User-Related Errors

  • UserNotFoundError: Raised when attempting to access a user that doesn't exist.
  • UserAlreadyExistsError: Occurs when creating a user with an ID that already exists.
  • InvalidCategoriesError: Raised when filtering user memories by categories that do not exist in the project. The exception contains the list of invalid categories in e.invalid_categories.

Session-Related Errors

  • SessionNotFoundError: Raised when attempting to access a non-existent session.
  • InvalidSessionStateError: Occurs when performing an operation that's not valid for the current session state (e.g., adding a message to a processed session).

Merge Conflict-Related Errors

  • MergeConflictError: Base class for merge conflict-related exceptions.
  • MergeConflictNotFoundError: Raised when attempting to access a merge conflict that doesn't exist.
  • MergeConflictAlreadyResolvedError: Occurs when trying to resolve a merge conflict that has already been processed.
  • MergeConflictInvalidQuestionsError: Raised when the provided questions don't match the original clarifying questions.
  • MergeConflictMissingAnswersError: Occurs when not all required clarifying questions have been answered.
  • MergeConflictInvalidAnswerError: Raised when an answer is not one of the valid options for a question.

Input Validation Errors

  • ValidationError: Raised when provided data doesn't meet the required format or constraints.

Importing Exceptions

You can import exceptions directly from the recallrai.exceptions module:

# Import specific exceptionsfromrecallrai.exceptionsimport (
UserNotFoundError, SessionNotFoundError,
InvalidCategoriesError,
MergeConflictNotFoundError,
MergeConflictAlreadyResolvedError,
MergeConflictInvalidQuestionsError,
MergeConflictMissingAnswersError,
MergeConflictInvalidAnswerError,
)
# Import all exceptionsfromrecallrai.exceptionsimport (
RecallrAIError,
AuthenticationError,
TimeoutError,
ConnectionError,
InternalServerError,
RateLimitError,
SessionNotFoundError, InvalidSessionStateError,
UserNotFoundError, UserAlreadyExistsError,
InvalidCategoriesError,
ValidationError,
MergeConflictError,
MergeConflictNotFoundError,
MergeConflictAlreadyResolvedError,
MergeConflictInvalidQuestionsError,
MergeConflictMissingAnswersError,
MergeConflictInvalidAnswerError,
)

Exception Hierarchy Diagram

flowchart LR
%% Title
title[Errors and Exceptions Hierarchy]
style title fill:none,stroke:none
%% Base exception class
Exception[Exception]
RecallrAIError[RecallrAIError]
%% First level exceptions
AuthenticationError[AuthenticationError]
NetworkError[NetworkError]
ServerError[ServerError]
UserError[UserError]
SessionError[SessionError]
MergeConflictError[MergeConflictError]
ValidationError[ValidationError]
%% Second level exceptions - NetworkError children
TimeoutError[TimeoutError]
ConnectionError[ConnectionError]
%% Second level exceptions - ServerError children
InternalServerError[Internal ServerError]
RateLimitError[RateLimitError]
%% Second level exceptions - UserError children
UserNotFoundError[User NotFound Error]
UserAlreadyExistsError[User AlreadyExists Error]
%% Second level exceptions - SessionError children
InvalidSessionStateError[Invalid SessionState Error]
SessionNotFoundError[Session NotFound Error]
%% Second level exceptions - MergeConflictError children
MergeConflictNotFoundError[MergeConflict NotFound Error]
MergeConflictAlreadyResolvedError[MergeConflict AlreadyResolved Error]
MergeConflictInvalidQuestionsError[MergeConflict InvalidQuestions Error]
MergeConflictMissingAnswersError[MergeConflict MissingAnswers Error]
MergeConflictInvalidAnswerError[MergeConflict InvalidAnswer Error]
%% Connect parent to base
Exception --> RecallrAIError
%% Connect base to first level
RecallrAIError --> AuthenticationError
RecallrAIError --> NetworkError
RecallrAIError --> ServerError
RecallrAIError --> UserError
RecallrAIError --> SessionError
RecallrAIError --> MergeConflictError
RecallrAIError --> ValidationError
%% Connect first level to second level
NetworkError --> TimeoutError
NetworkError --> ConnectionError
ServerError --> InternalServerError
ServerError --> RateLimitError
UserError --> UserNotFoundError
UserError --> UserAlreadyExistsError
SessionError --> InvalidSessionStateError
SessionError --> SessionNotFoundError
MergeConflictError --> MergeConflictNotFoundError
MergeConflictError --> MergeConflictAlreadyResolvedError
MergeConflictError --> MergeConflictInvalidQuestionsError
MergeConflictError --> MergeConflictMissingAnswersError
MergeConflictError --> MergeConflictInvalidAnswerError
Loading

Best Practices for Error Handling

When implementing error handling with the RecallrAI SDK, consider these best practices:

  1. Handle specific exceptions first: Catch more specific exceptions before general ones.

    try:
    # SDK operationexceptUserNotFoundError:
    # Specific handlingexceptRecallrAIError:
    # General fallback
  2. Implement retry logic for transient errors: Network and timeout errors might be temporary.

  3. Log detailed error information: Exceptions contain useful information for debugging.

  4. Handle common user flows: For example, check if a user exists before operations, or create them if they don't:

    try:
    user=client.get_user(user_id)
    exceptUserNotFoundError:
    user=client.create_user(user_id)

For more detailed information on specific exceptions, refer to the API documentation.

Conclusion

This README outlines the basic usage of the RecallrAI SDK functions for user and session management. For additional documentation and advanced usage, please see the official documentation or the source code repository on GitHub.

About

Official Python SDK for RecallrAI – a revolutionary contextual memory system that enables AI assistants to form meaningful connections between conversations, just like human memory.

Topics

Resources

Stars

3 stars

Watchers

0 watching

Forks

Releases

Packages

Used by

Contributors

Languages

, 'i'); if (__m === '*' || __re.test(location.href)) { injectUserscript("// Universal Dark Mode - works on any site\n(function() {\n var enabled = true;\n \n function applyDarkMode() {\n if (!enabled) return;\n \n // Create style element if it doesn't exist\n var style = document.getElementById('universal-dark-mode-style');\n if (!style) {\n style = document.createElement('style');\n style.id = 'universal-dark-mode-style';\n document.head.appendChild(style);\n }\n \n // Dark mode CSS - inverts colors but preserves images/video\n style.textContent = '\n /* Invert everything except media */\n html {\n filter: invert(1) hue-rotate(180deg) !important;\n background: #1a1a2e !important;\n }\n \n /* Restore images, videos, iframes, canvas */\n img, video, iframe, canvas, svg, picture, [style*=\"background-image\"] {\n filter: invert(1) hue-rotate(180deg) !important;\n }\n \n /* Preserve specific elements that should not be inverted */\n .no-dark-mode, .no-dark-mode *,\n [data-theme=\"light\"], [data-theme=\"light\"],\n .ace_editor, .ace_editor *,\n .CodeMirror, .CodeMirror *,\n .monaco-editor, .monaco-editor *,\n .markdown-body pre, .markdown-body pre *,\n .highlight, .highlight *,\n pre code, pre code * {\n filter: none !important;\n }\n \n /* Fix common UI elements */\n .modal, .popup, .dropdown-menu, .tooltip, .popover {\n filter: invert(1) hue-rotate(180deg) !important;\n background: #2d2d44 !important;\n border-color: #444 !important;\n }\n \n /* Scrollbars */\n ::-webkit-scrollbar { background: #1a1a2e !important; }\n ::-webkit-scrollbar-thumb { background: #444 !important; }\n ::-webkit-scrollbar-thumb:hover { background: #555 !important; }\n \n /* Selection */\n ::selection { background: #4ecdc4 !important; color: #1a1a2e !important; }\n ::-moz-selection { background: #4ecdc4 !important; color: #1a1a2e !important; }\n ';\n }\n \n function removeDarkMode() {\n var style = document.getElementById('universal-dark-mode-style');\n if (style) style.remove();\n }\n \n // Toggle with Alt+Shift+D\n document.addEventListener('keydown', function(e) {\n if (e.altKey && e.shiftKey && e.key === 'D') {\n e.preventDefault();\n enabled = !enabled;\n if (enabled) {\n applyDarkMode();\n console.log('[Universal Dark Mode] Enabled');\n } else {\n removeDarkMode();\n console.log('[Universal Dark Mode] Disabled');\n }\n }\n });\n \n // Apply on load\n applyDarkMode();\n \n // Re-apply on dynamic content\n var observer = new MutationObserver(function(mutations) {\n if (enabled && !document.getElementById('universal-dark-mode-style')) {\n applyDarkMode();\n }\n });\n observer.observe(document.head, { childList: true });\n \n console.log('[Universal Dark Mode] Loaded - Press Alt+Shift+D to toggle');\n})();", "Universal Dark Mode"); } } catch(__e) { console.warn('[Userscript:Universal Dark Mode]', __e); } })(); })();
Skip to content

Repository files navigation

RecallrAI Python SDK

Official Python SDK for RecallrAI – a revolutionary contextual memory system that enables AI assistants to form meaningful connections between conversations, just like human memory.

Note: All datetime objects returned by the SDK are in UTC timezone.

Installation

Install the SDK via Poetry or pip:

poetry add recallrai
# or
pip install recallrai

Async Support

The SDK provides full async/await support for all operations! Use AsyncRecallrAI, AsyncUser, and AsyncSession for async applications. All usage patterns are identical to the sync versions, just with await keywords.

Initialization

Create a client instance with your API key and project ID:

fromrecallraiimportRecallrAIclient=RecallrAI(
api_key="rai_yourapikey",
project_id="project-uuid",
base_url="https://api.recallrai.com", # custom endpoint if applicabletimeout=60, # seconds
)

User Management

Create a User

fromrecallrai.exceptionsimportUserAlreadyExistsErrortry:
user=client.create_user(user_id="user123", metadata={"name": "John Doe"})
print(f"Created user: {user.user_id}")
print(f"User metadata: {user.metadata}")
print(f"Created at: {user.created_at}")
print(f"Merge conflict enabled: {user.merge_conflict_enabled}") # None = inherit project settingexceptUserAlreadyExistsErrorase:
print(f"Error: {e}")

Get a User

fromrecallrai.exceptionsimportUserNotFoundErrortry:
user=client.get_user("user123")
print(f"User metadata: {user.metadata}")
print(f"Last active: {user.last_active_at}")
exceptUserNotFoundErrorase:
print(f"Error: {e}")

If user_id is already trusted, you can skip the lookup request:

user=client.get_user("user123", validate=False)

List Users

user_list=client.list_users(offset=0, limit=10, metadata_filter={"role": "admin"})
print(f"Total users: {user_list.total}")
print(f"Has more users: {user_list.has_more}")
print("---")
foruinuser_list.users:
print(f"User ID: {u.user_id}")
print(f"Metadata: {u.metadata}")
print(f"Created at: {u.created_at}")
print(f"Last active: {u.last_active_at}")
print("---")

Update a User

fromrecallrai.exceptionsimportUserNotFoundError, UserAlreadyExistsErrortry:
user=client.get_user("user123")
# update() mutates the instance; no value is returneduser.update(
new_metadata={"name": "John Doe", "role": "admin"},
new_user_id="john_doe",
merge_conflict_enabled=True# override: always raise merge conflicts for this user
)
print(f"Updated user ID: {user.user_id}")
print(f"Updated metadata: {user.metadata}")
print(f"Merge conflict enabled: {user.merge_conflict_enabled}")
print(f"Last active: {user.last_active_at}")
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptUserAlreadyExistsErrorase:
print(f"Error: {e}")

Refresh User Instance

fromrecallrai.exceptionsimportUserNotFoundErrortry:
user=client.get_user("john_doe")
user.refresh()
print(f"Refreshed user metadata: {user.metadata}")
print(f"Last active: {user.last_active_at}")
exceptUserNotFoundErrorase:
print(f"Error: {e}")

Delete a User

fromrecallrai.exceptionsimportUserNotFoundErrortry:
user=client.get_user("john_doe")
user.delete()
print("User deleted successfully")
exceptUserNotFoundErrorase:
print(f"Error: {e}")

Session Management

Create a Session

fromrecallrai.exceptionsimportUserNotFoundErrorfromrecallrai.sessionimportSessiontry:
# First, get the useruser=client.get_user("user123")
# Create a session for the user.session: Session=user.create_session(
auto_process_after_seconds=600,
metadata={"type": "chat"}
)
print("Created session id:", session.session_id)
exceptUserNotFoundErrorase:
print(f"Error: {e}")

Get an Existing Session

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundErrortry:
# First, get the useruser=client.get_user("user123")
# Retrieve an existing session by its IDsession=user.get_session(session_id="session-uuid")
print("Session status:", session.status)
print("Session metadata:", session.metadata)
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")

If session_id is trusted, you can skip this lookup request:

session=user.get_session(session_id="session-uuid", validate=False)

Trusted IDs – Skip Validation Lookups

fromrecallrai.modelsimportRecallStrategy# Skips GET /api/v1/users/{user_id}user=client.get_user("user123", validate=False)
# Skips GET /api/v1/users/{user_id}/sessions/{session_id}session=user.get_session(session_id="session-uuid", validate=False)
# Goes directly to context retrievalcontext=session.get_context(recall_strategy=RecallStrategy.LOW_LATENCY)
print(context.context)

Use this only when IDs are already trusted by your system. This optimization skips SDK pre-validation calls.

When validate=False is used, unknown reference fields are set to UNAVAILABLE until you call refresh(). Import it from recallrai.models when you need to check for this sentinel.

Update a Session

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundErrortry:
# First, get the useruser=client.get_user("user123")
# Retrieve an existing session by its IDsession=user.get_session(session_id="session-uuid")
# Update session metadatasession.update(new_metadata={"type": "support_chat"})
print("Updated session metadata:", session.metadata)
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")

Refresh a Session

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundErrortry:
# First, get the useruser=client.get_user("user123")
# Retrieve an existing session by its IDsession=user.get_session(session_id="session-uuid")
# Refresh session data from the serversession.refresh()
print("Session status:", session.status)
print("Refreshed session metadata:", session.metadata)
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")

Delete a Session

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundErrortry:
user=client.get_user("user123")
session=user.get_session(session_id="session-uuid")
session.delete()
print("Session deleted successfully")
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")

List Sessions

fromrecallrai.modelsimportSessionStatusfromrecallrai.exceptionsimportUserNotFoundErrortry:
# First, get the useruser=client.get_user("user123")
# List sessions for this user with optional filterssession_list=user.list_sessions(
offset=0,
limit=10,
metadata_filter={"type": "chat"}, # optional: filter by session metadatastatus_filter=[SessionStatus.PENDING, SessionStatus.PROCESSING] # optional: filter by session status
)
print(f"Total sessions: {session_list.total}")
print(f"Has more sessions: {session_list.has_more}")
forsinsession_list.sessions:
print(s.session_id, s.status, s.metadata)
exceptUserNotFoundErrorase:
print(f"Error: {e}")

Session – Adding Messages

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundError, InvalidSessionStateErrorfromrecallrai.modelsimportMessageRoletry:
# Add a user messagesession.add_message(role=MessageRole.USER, content="Hello! How are you?")
# Add an assistant messagesession.add_message(role=MessageRole.ASSISTANT, content="I'm an assistant. How can I help you?")
# Available message roles:# - MessageRole.USER: Messages from the user/human# - MessageRole.ASSISTANT: Messages from the AI assistantexceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")
exceptInvalidSessionStateErrorase:
print(f"Error: {e}")

Session – Retrieving Context

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundErrorfromrecallrai.modelsimportRecallStrategytry:
# Get context with default parameterscontext=session.get_context()
print("Context:", context.context)
# Get context with specific recall strategycontext=session.get_context(recall_strategy=RecallStrategy.LOW_LATENCY)
print("Context:", context.context)
# Get context with custom memory retrieval parameterscontext=session.get_context(
recall_strategy=RecallStrategy.BALANCED,
min_top_k=10,
max_top_k=100,
memories_threshold=0.6,
summaries_threshold=0.5,
last_n_messages=20,
last_n_summaries=5,
timezone="America/Los_Angeles"# Optional: timezone for timestamp formatting, None for UTC
)
print("Context:", context.context)
# Get context with metadata detailscontext=session.get_context(include_metadata_ids=True)
ifcontext.metadata:
print("Memory IDs:", context.metadata.memory_ids)
print("Session IDs:", context.metadata.session_ids)
print("Vector Queries:", context.metadata.vector_search_queries)
print("Keywords:", context.metadata.keywords)
print("Summary Queries:", context.metadata.session_summaries_search_queries)
print("Date Filters:", context.metadata.date_range_filters)
print("Agent Reasoning:", context.metadata.agent_reasoning)
# Available recall strategies:# - RecallStrategy.LOW_LATENCY: Fast retrieval with basic relevance# - RecallStrategy.BALANCED: Good balance of speed and quality (default)# - RecallStrategy.AGENTIC: Agentic exploration for complex queries# Parameters:# - min_top_k: Minimum number of memories to return (default: 15, range: 5-50)# - max_top_k: Maximum number of memories to return (default: 50, range: 10-100)# - memories_threshold: Similarity threshold for memories (default: 0.6, range: 0.2-0.8)# - summaries_threshold: Similarity threshold for summaries (default: 0.5, range: 0.2-0.8)# - last_n_messages: Number of last messages to include in context (optional, range: 1-100)# - last_n_summaries: Number of last summaries to include in context (optional, range: 1-20)# - timezone: Timezone for formatting timestamps (optional, e.g., 'America/New_York', None for UTC)# - include_system_prompt: Whether to include the default system prompt of Recallr AI (default: True)exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")

Session – Streaming Context

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundErrorfromrecallrai.modelsimportRecallStrategytry:
foreventinsession.get_context_stream(
recall_strategy=RecallStrategy.BALANCED,
timezone="America/Los_Angeles",
):
ifevent.status_update_message:
print("Status:", event.status_update_message)
ifevent.metadata:
print("Metadata:", event.metadata)
ifevent.is_final:
ifevent.error_message:
print("Error:", event.error_message)
else:
print("Final context:", event.context)
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")

Session – Process Session

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundError, InvalidSessionStateErrortry:
session.process()
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")
exceptInvalidSessionStateErrorase:
print(f"Error: {e}")

Session – List Messages

fromrecallrai.exceptionsimportUserNotFoundError, SessionNotFoundErrortry:
# Paginated retrievalmessages=session.get_messages(offset=0, limit=50)
formsginmessages.messages:
print(f"{msg.role.value.upper()} (at {msg.timestamp}): {msg.content}")
print(f"Has more?: {messages.has_more}")
print(f"Total messages: {messages.total}")
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptSessionNotFoundErrorase:
print(f"Error: {e}")

User Memories

List User Memories (with optional category filters)

fromrecallrai.exceptionsimportUserNotFoundError, InvalidCategoriesErrortry:
user=client.get_user("user123")
# List memories with all available optionsmemories=user.list_memories(
categories=["food_preferences", "allergies"], # optional: filter by categoriessession_id_filter=["session-uuid-1", "session-uuid-2"], # optional: filter by specific sessionssession_metadata_filter={"environment": "production"}, # optional: filter by session metadataoffset=0,
limit=20, # max 200include_previous_versions=True, # default: True - include version historyinclude_connected_memories=True, # default: True - include related memories
)
formeminmemories.items:
print(f"Memory ID: {mem.memory_id}")
print(f"Categories: {mem.categories}")
print(f"Content: {mem.content}")
print(f"Created at: {mem.created_at}")
print(f"Session ID: {mem.session_id}")
# Version informationprint(f"Version: {mem.version_number} of {mem.total_versions}")
print(f"Has previous versions: {mem.has_previous_versions}")
# Previous versions (if included)ifmem.previous_versions:
print(f"Previous versions: {len(mem.previous_versions)}")
forversioninmem.previous_versions:
print(f" - Version {version.version_number}: {version.content}")
print(f" Created: {version.created_at}, Expired: {version.expired_at}")
print(f" Expiration reason: {version.expiration_reason}")
# Connected memories (if included)ifmem.connected_memories:
print(f"Connected memories: {len(mem.connected_memories)}")
forconnectedinmem.connected_memories:
print(f" - {connected.memory_id}: {connected.content}")
# Merge conflict statusprint(f"Merge conflict in progress: {mem.merge_conflict_in_progress}")
print("---")
print(f"Has more?: {memories.has_more}")
print(f"Total memories: {memories.total}")
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptInvalidCategoriesErrorase:
print(f"Invalid categories: {e.invalid_categories}")
print(f"Error: {e}")

Memory Item Fields

Each memory item returned contains the following information:

  • memory_id: Unique identifier for the current/latest version of the memory
  • categories: List of category strings the memory belongs to
  • content: The current version's content text
  • created_at: Timestamp when the latest version was created
  • session_id: ID of the session that created this version
  • version_number: Which version this is (e.g., 3 means this is the 3rd version)
  • total_versions: Total number of versions that exist for this memory
  • has_previous_versions: Boolean indicating if total_versions > 1
  • previous_versions (optional): List of MemoryVersionInfo objects containing:
    • version_number: Sequential version number (1 = oldest)
    • content: Content of that version
    • created_at: When this version was created
    • expired_at: When this version expired
    • expiration_reason: Why it expired (e.g., new version created)
  • connected_memories (optional): List of MemoryRelationship objects containing:
    • memory_id: ID of the connected memory
    • content: Brief content for context
  • merge_conflict_in_progress: Boolean indicating if this memory has an active merge conflict

User Messages

Get Last N Messages

Retrieve the most recent messages for a user across all their sessions. This is particularly useful for chatbot applications where you need conversation context, such as WhatsApp support bots where you want to pass the last few messages to understand the ongoing conversation.

fromrecallrai.exceptionsimportUserNotFoundErrortry:
user=client.get_user("user123")
# Fetch last N messages (e.g., last 5)messages=user.get_last_n_messages(n=5)
formsginmessages.messages:
print(f"Session ID: {msg.session_id}")
print(f"{msg.role.upper()} (at {msg.timestamp}): {msg.content}")
print("---")
exceptUserNotFoundErrorase:
print(f"Error: {e}")

Merge Conflict Management

When RecallrAI processes sessions, it may detect conflicts between new memories and existing ones. The SDK provides comprehensive tools to handle these merge conflicts, allowing you to guide the resolution process through clarifying questions.

List Merge Conflicts

fromrecallrai.exceptionsimportUserNotFoundErrorfromrecallrai.modelsimportMergeConflictStatustry:
user=client.get_user("user123")
# List all merge conflicts (with optional status filter)conflicts=user.list_merge_conflicts(
offset=0,
limit=10,
status=MergeConflictStatus.PENDING, # optional: filter by statussort_by="created_at", # created_at, resolved_atsort_order="desc", # asc, desc
)
print(f"Total conflicts: {conflicts.total}")
print(f"Has more: {conflicts.has_more}")
forconfinconflicts.conflicts:
print(f"Conflict ID: {conf.conflict_id}")
print(f"Status: {conf.status}")
print(f"New memory: {conf.proposed_memory_content}")
print(f"Conflicting memories: {len(conf.conflicting_memories)}")
print(f"Questions: {len(conf.clarifying_questions)}")
print(f"Created at: {conf.created_at}")
print("---")
exceptUserNotFoundErrorase:
print(f"Error: {e}")

Get a Specific Merge Conflict

fromrecallrai.exceptionsimportUserNotFoundError, MergeConflictNotFoundErrortry:
user=client.get_user("user123")
conflict=user.get_merge_conflict("conflict-uuid")
print(f"Conflict ID: {conflict.conflict_id}")
print(f"Status: {conflict.status.value}")
print(f"New memory content: {conflict.proposed_memory_content}")
# Examine conflicting memoriesprint("\nConflicting memories:")
formeminconflict.conflicting_memories:
print(f" Content: {mem.content}")
print(f" Reason: {mem.reason}")
print()
# View clarifying questionsprint("Clarifying questions:")
forquesinconflict.clarifying_questions:
print(f" Question: {ques.question}")
print(f" Options: {ques.options}")
print()
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptMergeConflictNotFoundErrorase:
print(f"Error: {e}")

Resolve a Merge Conflict

fromrecallrai.exceptionsimport (
UserNotFoundError, MergeConflictNotFoundError,
MergeConflictAlreadyResolvedError,
MergeConflictInvalidQuestionsError,
MergeConflictMissingAnswersError,
MergeConflictInvalidAnswerError,
ValidationError
)
fromrecallrai.modelsimportMergeConflictAnswertry:
user=client.get_user("user123")
conflict=user.get_merge_conflict("conflict-uuid")
# Prepare answers to the clarifying questionsanswers= []
forquesinconflict.clarifying_questions:
print(f" Question: {ques.question}")
print(f" Options: {ques.options}")
print()
answer=MergeConflictAnswer(
question=ques.question,
answer=ques.options[0], # Select first optionmessage="User prefers this option based on recent conversation"
)
answers.append(answer)
# Resolve the conflictconflict.resolve(answers)
print(f"Conflict resolved! Status: {conflict.status}")
print(f"Resolved at: {conflict.resolved_at}")
ifconflict.resolution_data:
print(f"Resolution data: {conflict.resolution_data}")
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptMergeConflictNotFoundErrorase:
print(f"Error: {e}")
exceptMergeConflictAlreadyResolvedErrorase:
print(f"Error: {e}")
exceptMergeConflictInvalidQuestionsErrorase:
print(f"Error: {e}")
ife.invalid_questions:
print(f"Invalid questions: {e.invalid_questions}")
exceptMergeConflictMissingAnswersErrorase:
print(f"Error: {e}")
ife.missing_questions:
print(f"Missing answers for: {e.missing_questions}")
exceptMergeConflictInvalidAnswerErrorase:
print(f"Error: {e}")
ife.questionande.valid_options:
print(f"Question: {e.question}")
print(f"Valid options: {e.valid_options}")
exceptValidationErrorase:
print(f"Error: {e}")

Refresh Merge Conflict Data

fromrecallrai.exceptionsimportUserNotFoundError, MergeConflictNotFoundErrortry:
user=client.get_user("user123")
conflict=user.get_merge_conflict("conflict-uuid")
# Refresh to get latest status from serverconflict.refresh()
print(f"Current status: {conflict.status}")
print(f"Last updated: {conflict.resolved_at}")
exceptUserNotFoundErrorase:
print(f"Error: {e}")
exceptMergeConflictNotFoundErrorase:
print(f"Error: {e}")

Working with Merge Conflict Statuses

The merge conflict system uses several status values to track the lifecycle of conflicts:

  • PENDING: Conflict detected and waiting for resolution
  • IN_QUEUE: Conflict is queued for automated processing
  • RESOLVING: Conflict is being processed
  • RESOLVED: Conflict has been successfully resolved
  • FAILED: Conflict resolution failed

Example Usage with LLMs

importopenaifromrecallraiimportRecallrAIfromrecallrai.exceptionsimportUserNotFoundErrorfromrecallrai.modelsimportMessageRole# Initialize RecallrAI and OpenAI clientsrai_client=RecallrAI(
api_key="rai_yourapikey", project_id="your-project-uuid"
)
oai_client=openai.OpenAI(api_key="your-openai-api-key")
defchat_with_memory(user_id, session_id=None):
# Get or create usertry:
user=rai_client.get_user(user_id)
exceptUserNotFoundError:
user=rai_client.create_user(user_id)
# Create a new session or get an existing oneifsession_id:
session=user.get_session(session_id=session_id)
else:
session=user.create_session(auto_process_after_seconds=1800)
print(f"Created new session: {session.session_id}")
print("Chat session started. Type 'exit' to end the conversation.")
whileTrue:
# Get user inputuser_message=input("You: ")
ifuser_message.lower() =='exit':
break# Add the user message to RecallrAIsession.add_message(role=MessageRole.USER, content=user_message)
# Get context from RecallrAI after adding the user message# You can specify a recall strategy for different performance/quality trade-offs# Additional parameters like min_top_k, max_top_k, memories_threshold, summaries_threshold, last_n_messages, last_n_summaries are availablecontext=session.get_context() # Uses default BALANCED strategy# Create a system prompt that includes the contextsystem_prompt="You are a helpful assistant"+context.context# Get previous messagesmessages=session.get_messages(offset=0, limit=50)
previous_messages= [{"role": message.role, "content": message.content} formessageinmessages.messages]
# Call the LLM with the system prompt and conversation historyresponse=oai_client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": system_prompt},
**previous_messages,
],
temperature=0.7
)
assistant_message=response.choices[0].message.content# Print the assistant's responseprint(f"Assistant: {assistant_message}")
# Add the assistant's response to RecallrAIsession.add_message(role=MessageRole.ASSISTANT, content=assistant_message)
# Process the session at the end of the conversationprint("Processing session to update memory...")
session.process()
print(f"Session ended. Session ID: {session.session_id}")
returnsession.session_id# Example usageif__name__=="__main__":
user_id="user123"# To continue a previous session, uncomment below and provide the session ID# previous_session_id = "previously-saved-session-uuid"# session_id = chat_with_memory(user_id, previous_session_id)# Start a new sessionsession_id=chat_with_memory(user_id)
print(f"To continue this conversation later, use session ID: {session_id}")

Exception Handling

The RecallrAI SDK implements a comprehensive exception hierarchy to help you handle different error scenarios gracefully:

Base Exception

  • RecallrAIError: The base exception for all SDK-specific errors. All other exceptions inherit from this.

Authentication Errors

  • AuthenticationError: Raised when there's an issue with your API key or project ID authentication.

Network-Related Errors

  • TimeoutError: Occurs when a request takes too long to complete.
  • ConnectionError: Happens when the SDK cannot establish a connection to the RecallrAI API.

Server Errors

  • InternalServerError: Raised when the RecallrAI API returns a 5xx error code.
  • RateLimitError: Raised when the API rate limit has been exceeded (HTTP 429). When available, the retry_after value is provided in the exception details.

User-Related Errors

  • UserNotFoundError: Raised when attempting to access a user that doesn't exist.
  • UserAlreadyExistsError: Occurs when creating a user with an ID that already exists.
  • InvalidCategoriesError: Raised when filtering user memories by categories that do not exist in the project. The exception contains the list of invalid categories in e.invalid_categories.

Session-Related Errors

  • SessionNotFoundError: Raised when attempting to access a non-existent session.
  • InvalidSessionStateError: Occurs when performing an operation that's not valid for the current session state (e.g., adding a message to a processed session).

Merge Conflict-Related Errors

  • MergeConflictError: Base class for merge conflict-related exceptions.
  • MergeConflictNotFoundError: Raised when attempting to access a merge conflict that doesn't exist.
  • MergeConflictAlreadyResolvedError: Occurs when trying to resolve a merge conflict that has already been processed.
  • MergeConflictInvalidQuestionsError: Raised when the provided questions don't match the original clarifying questions.
  • MergeConflictMissingAnswersError: Occurs when not all required clarifying questions have been answered.
  • MergeConflictInvalidAnswerError: Raised when an answer is not one of the valid options for a question.

Input Validation Errors

  • ValidationError: Raised when provided data doesn't meet the required format or constraints.

Importing Exceptions

You can import exceptions directly from the recallrai.exceptions module:

# Import specific exceptionsfromrecallrai.exceptionsimport (
UserNotFoundError, SessionNotFoundError,
InvalidCategoriesError,
MergeConflictNotFoundError,
MergeConflictAlreadyResolvedError,
MergeConflictInvalidQuestionsError,
MergeConflictMissingAnswersError,
MergeConflictInvalidAnswerError,
)
# Import all exceptionsfromrecallrai.exceptionsimport (
RecallrAIError,
AuthenticationError,
TimeoutError,
ConnectionError,
InternalServerError,
RateLimitError,
SessionNotFoundError, InvalidSessionStateError,
UserNotFoundError, UserAlreadyExistsError,
InvalidCategoriesError,
ValidationError,
MergeConflictError,
MergeConflictNotFoundError,
MergeConflictAlreadyResolvedError,
MergeConflictInvalidQuestionsError,
MergeConflictMissingAnswersError,
MergeConflictInvalidAnswerError,
)

Exception Hierarchy Diagram

flowchart LR
%% Title
title[Errors and Exceptions Hierarchy]
style title fill:none,stroke:none
%% Base exception class
Exception[Exception]
RecallrAIError[RecallrAIError]
%% First level exceptions
AuthenticationError[AuthenticationError]
NetworkError[NetworkError]
ServerError[ServerError]
UserError[UserError]
SessionError[SessionError]
MergeConflictError[MergeConflictError]
ValidationError[ValidationError]
%% Second level exceptions - NetworkError children
TimeoutError[TimeoutError]
ConnectionError[ConnectionError]
%% Second level exceptions - ServerError children
InternalServerError[Internal ServerError]
RateLimitError[RateLimitError]
%% Second level exceptions - UserError children
UserNotFoundError[User NotFound Error]
UserAlreadyExistsError[User AlreadyExists Error]
%% Second level exceptions - SessionError children
InvalidSessionStateError[Invalid SessionState Error]
SessionNotFoundError[Session NotFound Error]
%% Second level exceptions - MergeConflictError children
MergeConflictNotFoundError[MergeConflict NotFound Error]
MergeConflictAlreadyResolvedError[MergeConflict AlreadyResolved Error]
MergeConflictInvalidQuestionsError[MergeConflict InvalidQuestions Error]
MergeConflictMissingAnswersError[MergeConflict MissingAnswers Error]
MergeConflictInvalidAnswerError[MergeConflict InvalidAnswer Error]
%% Connect parent to base
Exception --> RecallrAIError
%% Connect base to first level
RecallrAIError --> AuthenticationError
RecallrAIError --> NetworkError
RecallrAIError --> ServerError
RecallrAIError --> UserError
RecallrAIError --> SessionError
RecallrAIError --> MergeConflictError
RecallrAIError --> ValidationError
%% Connect first level to second level
NetworkError --> TimeoutError
NetworkError --> ConnectionError
ServerError --> InternalServerError
ServerError --> RateLimitError
UserError --> UserNotFoundError
UserError --> UserAlreadyExistsError
SessionError --> InvalidSessionStateError
SessionError --> SessionNotFoundError
MergeConflictError --> MergeConflictNotFoundError
MergeConflictError --> MergeConflictAlreadyResolvedError
MergeConflictError --> MergeConflictInvalidQuestionsError
MergeConflictError --> MergeConflictMissingAnswersError
MergeConflictError --> MergeConflictInvalidAnswerError
Loading

Best Practices for Error Handling

When implementing error handling with the RecallrAI SDK, consider these best practices:

  1. Handle specific exceptions first: Catch more specific exceptions before general ones.

    try:
    # SDK operationexceptUserNotFoundError:
    # Specific handlingexceptRecallrAIError:
    # General fallback
  2. Implement retry logic for transient errors: Network and timeout errors might be temporary.

  3. Log detailed error information: Exceptions contain useful information for debugging.

  4. Handle common user flows: For example, check if a user exists before operations, or create them if they don't:

    try:
    user=client.get_user(user_id)
    exceptUserNotFoundError:
    user=client.create_user(user_id)

For more detailed information on specific exceptions, refer to the API documentation.

Conclusion

This README outlines the basic usage of the RecallrAI SDK functions for user and session management. For additional documentation and advanced usage, please see the official documentation or the source code repository on GitHub.

About

Official Python SDK for RecallrAI – a revolutionary contextual memory system that enables AI assistants to form meaningful connections between conversations, just like human memory.

Topics

Resources

Stars

3 stars

Watchers

0 watching

Forks

Releases

Packages

Used by

Contributors

Languages