PiyAPI Memory Python SDK

Official Python SDK for PiyAPI Memory - AI-native memory and context management for your applications.

Installation

pip install piyapi-memory

For async support:

pip install piyapi-memory[async]

Quick Start

from piyapi_memory import PiyAPIClient

# Initialize the client
client = PiyAPIClient(api_key="your-api-key")

# Create a memory
memory = client.create_memory(
    content="The user prefers dark mode and uses Python primarily.",
    metadata={"category": "preferences", "source": "onboarding"},
    namespace="user-123"
)

# Search memories
results = client.semantic_search(
    query="What are the user's preferences?",
    namespace="user-123",
    limit=5
)

for result in results:
    print(f"Score: {result.score:.3f} - {result.content}")

Async Usage

import asyncio
from piyapi_memory import AsyncPiyAPIClient

async def main():
    client = AsyncPiyAPIClient(api_key="your-api-key")
    
    # Create a memory
    memory = await client.create_memory(
        content="User completed the tutorial",
        namespace="user-123"
    )
    
    # Search
    results = await client.semantic_search(
        query="tutorial progress",
        namespace="user-123"
    )
    
    await client.close()

asyncio.run(main())

Features

• Memory CRUD: Create, read, update, and delete memories
• Semantic Search: Find relevant memories using natural language
• Keyword Search: Traditional text-based search
• Hybrid Search: Combine semantic and keyword search
• Namespaces: Organize memories by user, project, or any grouping
• Metadata Filtering: Filter search results by metadata
• Automatic Retry: Built-in retry with exponential backoff for rate limits
• Type Safety: Full type hints and Pydantic models

API Reference

Client Initialization

from piyapi_memory import PiyAPIClient

client = PiyAPIClient(
    api_key="your-api-key",
    base_url="https://api.piyapi.cloud",  # Optional, defaults to production
    timeout=30.0,  # Request timeout in seconds
    max_retries=3,  # Max retries on rate limit
)

Memory Operations

# Create
memory = client.create_memory(
    content="Memory content",
    metadata={"key": "value"},  # Optional
    namespace="default"  # Optional
)

# Get
memory = client.get_memory(memory_id="mem_123")

# Update
memory = client.update_memory(
    memory_id="mem_123",
    content="Updated content",
    metadata={"updated": True}
)

# Delete
client.delete_memory(memory_id="mem_123")

# List
memories = client.list_memories(
    namespace="user-123",
    limit=20,
    cursor="cursor_token"  # For pagination
)

Search Operations

# Semantic search
results = client.semantic_search(
    query="natural language query",
    namespace="user-123",
    limit=10,
    threshold=0.7,  # Minimum similarity score
    metadata_filter={"category": "preferences"}
)

# Keyword search
results = client.keyword_search(
    query="exact keywords",
    namespace="user-123",
    limit=10
)

# Hybrid search (combines semantic + keyword)
results = client.hybrid_search(
    query="search query",
    namespace="user-123",
    limit=10,
    semantic_weight=0.7  # 0-1, weight for semantic vs keyword
)

Memory Versioning

Track changes and rollback to previous versions:

# Get version history
history = client.get_versions("mem_123", limit=10)
print(f"Total versions: {history['totalVersions']}")
for version in history["versions"]:
    print(f"Version {version['version']}: {version['content'][:50]}...")

# Get specific version
version2 = client.get_version("mem_123", 2)
print(f"Version 2 content: {version2['version']['content']}")

# Rollback to previous version
result = client.rollback("mem_123", 2)
print(f"Rolled back to version {result['rollback']['restoredVersion']}")

Related Memories

Get parent-child memory relationships:

# Get related memories
related = client.get_related("mem_123", direction="both", depth=2)

if related.get("parent"):
    print(f"Parent: {related['parent']['content']}")
print(f"Children: {len(related['children'])}")

Error Handling

The SDK raises typed exceptions for different error conditions:

from piyapi_memory import PiyAPIClient
from piyapi_memory.exceptions import (
    AuthenticationError,
    RateLimitError,
    ValidationError,
    NotFoundError,
    ServerError,
)

client = PiyAPIClient(api_key="your-api-key")

try:
    memory = client.create_memory(content="test")
except AuthenticationError:
    print("Invalid API key")
except RateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after} seconds")
except ValidationError as e:
    print(f"Invalid request: {e.message}")
except NotFoundError:
    print("Resource not found")
except ServerError:
    print("Server error, please retry")

Configuration

Environment Variables

PIYAPI_API_KEY=your-api-key
PIYAPI_BASE_URL=https://api.piyapi.cloud  # Optional

import os
from piyapi_memory import PiyAPIClient

# Client will use PIYAPI_API_KEY from environment
client = PiyAPIClient()

License

MIT License - see LICENSE for details.