Python SDK
The Papr Python SDK provides a Pythonic interface for interacting with the Papr Memory API, with full type hints and modern Python features.
Installation
Install the SDK using pip:
pip install papr-memoryRequirements
- HTTPS: The endpoint requires HTTPS protocol. HTTP connections will fail.
- Environment Variable: Set
PAPR_MEMORY_API_KEYin your environment before running. - X-Client-Type Header: This is automatically included by the SDK but can be customized if needed.
- Authentication: Only use one auth method (X-API-Key takes precedence over Bearer token).
Quick Start
from papr_memory import Papr
import os
# Initialize the client
client = Papr(
x_api_key=os.environ.get("PAPR_MEMORY_API_KEY") # Updated parameter name
)
# Add a memory
memory = client.memory.add(
content="Important meeting notes from today's discussion",
type="text",
metadata={
"topics": ["meetings", "notes"],
"createdAt": "2024-01-01T12:00:00Z"
}
)
# Search memories with agentic graph enabled
results = client.memory.search(
query="meeting notes from today",
enable_agentic_graph=True, # Recommended for better results
rank_results=False
)
# Add memories in batch
memories = client.memory.add_batch(
memories=[
{
"content": "First memory",
"type": "text",
"metadata": {"topics": ["topic1"]}
},
{
"content": "Second memory",
"type": "text",
"metadata": {"topics": ["topic2"]}
}
]
)Authentication Methods
The SDK supports three authentication methods:
# API Key Authentication (recommended for most scenarios)
client = Papr(x_api_key="your-api-key") # Updated parameter name
# Bearer Token Authentication
client = Papr(bearer_token="your-bearer-token")
# Session Token Authentication (for browser-based apps)
client = Papr(session_token="your-session-token")Key SDK Features
- Full type hints support
- Automatic retries with exponential backoff
- Comprehensive error handling
- Async support
- Extensive documentation and examples
Main API Operations
# Memory Operations
client.memory.add(content="Memory content", type="text")
client.memory.get("memory_id")
client.memory.update("memory_id", content="Updated content")
client.memory.delete("memory_id")
client.memory.search(query="Search query", enable_agentic_graph=True)
client.memory.add_batch(memories=[...])
# User Operations
client.user.create(external_id="user123")
client.user.get("user_id")
client.user.update("user_id", email="updated@example.com")
client.user.delete("user_id")
client.user.list()
client.user.create_batch(users=[...])
# Messages Operations (NEW)
client.messages.store(
content="User message",
role="user",
session_id="session_123",
external_user_id="user_456",
process_messages=True
)
client.messages.get_history(session_id="session_123", limit=50)
client.messages.compress_session(session_id="session_123")
client.messages.get_session_status(session_id="session_123")
client.messages.process_session(session_id="session_123")
# Feedback Operations
client.feedback.submit(
search_id="search_id",
feedback_data={
"feedback_type": "thumbs_up",
"feedback_source": "inline"
}
)
client.feedback.get_by_id("feedback_id")
client.feedback.submit_batch(feedback_items=[...])Documentation
For complete documentation, examples, and API reference, visit our Python SDK repository.
Error Handling
The SDK provides typed exceptions for different error cases:
from papr_memory import Papr, APIError, NetworkError
client = Papr(x_api_key="your-api-key") # Updated parameter name
try:
client.memory.add(
content="Test memory",
type="text"
)
except APIError as e:
# Handle API errors (400, 401, 403, etc.)
print(f"API error {e.status_code}: {e.message}")
except NetworkError as e:
# Handle network errors
print(f"Network error: {e}")Advanced Usage
Using Memory Policies
Control knowledge graph generation with memory_policy:
# Auto mode with schema
response = client.memory.add(
content="Meeting with Jane Smith from Acme Corp about Q4 project",
memory_policy={
"mode": "auto",
"schema_id": "crm_schema"
}
)
# Auto mode with node constraints
response = client.memory.add(
content="Customer order for product SKU-123",
memory_policy={
"mode": "auto",
"schema_id": "ecommerce_schema",
"node_constraints": [
{
"node_type": "Product",
"create": "never", # Only link to existing products
"search": {
"properties": [{"name": "sku", "mode": "exact"}]
}
}
]
}
)
# Manual mode with exact nodes
response = client.memory.add(
content="Signed contract with Acme Corp for $50,000",
memory_policy={
"mode": "manual",
"nodes": [
{
"id": "contract_acme_2024",
"type": "Contract",
"properties": {
"title": "Service Agreement 2024",
"value": 50000,
"status": "active"
}
}
],
"relationships": [
{
"source": "$this",
"target": "contract_acme_2024",
"type": "REFERS_TO"
}
]
}
)
# With ACL and consent (OMO standard)
response = client.memory.add(
content="Sensitive customer data",
memory_policy={
"mode": "auto",
"consent": "explicit",
"risk": "sensitive",
"acl": {
"read": ["external_user:alice"],
"write": ["external_user:alice"]
}
}
)Messages API
Store and manage conversation history:
# Store a user message
message_response = client.messages.store(
content="How do I integrate Papr with my app?",
role="user",
session_id="conv_help_001",
title="Integration Help",
external_user_id="user_789",
process_messages=True # Create memories automatically
)
# Store assistant response
client.messages.store(
content="To integrate Papr, first install the SDK...",
role="assistant",
session_id="conv_help_001",
process_messages=True # Capture learnings
)
# Get conversation history
history = client.messages.get_history(
session_id="conv_help_001",
limit=50,
skip=0
)
print(f"Total messages: {history.total_count}")
for msg in history.messages:
print(f"{msg.role}: {msg.content}")
# Compress long conversation
compressed = client.messages.compress_session(
session_id="conv_help_001"
)
# Use compressed context in your LLM prompt
llm_context = compressed.context_for_llm
print(f"Compressed to {len(llm_context)} characters")
# Check session status
status = client.messages.get_session_status(
session_id="conv_help_001"
)
print(f"Session: {status.title}")
print(f"Messages: {status.message_count}")
print(f"Memories created: {status.memories_created}")Async Support
from papr_memory import AsyncPapr
async with AsyncPapr(x_api_key="your-api-key") as client: # Updated parameter name
memory = await client.memory.add(
content="Async memory creation",
type="text"
)Batch Operations
memories = client.memory.add_batch(
memories=[
{
"content": "Memory 1",
"type": "text",
"metadata": {"topics": ["topic1"]}
},
{
"content": "Memory 2",
"type": "text",
"metadata": {"topics": ["topic2"]}
}
],
batch_size=10 # Optional: number of items to process in parallel
)Search with Agentic Graph
# HIGHLY RECOMMENDED: Enable agentic graph for intelligent context-aware search
results = client.memory.search(
query="Find discussions about API performance issues from last month",
enable_agentic_graph=True, # Enables intelligent entity relationship navigation
max_memories=20, # Recommended for comprehensive memory coverage
max_nodes=15 # Recommended for comprehensive graph entity relationships
)
# Access both memories and graph nodes
for memory in results.data.memories:
print(f"Memory: {memory.content}")
# Work with graph nodes if available
for node in results.data.nodes:
print(f"Node: {node.label} - {node.properties.get('name', node.properties.get('id'))}")Submitting Feedback
# Submit feedback for a search result
feedback_response = client.feedback.submit(
search_id="search_123",
feedback_data={
"feedback_type": "thumbs_up",
"feedback_source": "inline",
"feedback_text": "This result was exactly what I needed"
}
)
# Submit batch feedback
batch_response = client.feedback.submit_batch(
feedback_items=[
{
"search_id": "search_123",
"feedback_data": {
"feedback_type": "thumbs_up",
"feedback_source": "inline"
}
},
{
"search_id": "search_456",
"feedback_data": {
"feedback_type": "memory_relevance",
"feedback_source": "memory_citation",
"feedback_score": 0.9
}
}
]
)