Python SDK

pip install goodmem-client

uv pip install goodmem-client

import goodmem_client

python setup.py install --user

import goodmem_client

## Import required modules
from goodmem_client.api import APIKeysApi, SpacesApi
from goodmem_client.configuration import Configuration
from goodmem_client.api_client import ApiClient
from goodmem_client.models import CreateApiKeyRequest
from goodmem_client.streaming import MemoryStreamClient
from goodmem_client.rest import ApiException
from pprint import pprint

## Configure the API client
configuration = Configuration()
configuration.host = "http://localhost:8080"  # Use your server URL

## Set authentication
configuration.api_key = {"ApiKeyAuth": "your-api-key"}

## Create API client
api_client = ApiClient(configuration=configuration)  # Your API key here

## Create an instance of the API class
api_instance = APIKeysApi(api_client=api_client)

## Prepare a create request with labels
create_request = CreateApiKeyRequest(
    labels={
        "environment": "development",
        "service": "chat-ui"
    }
)

try:
    # Create a new API key
    api_response = api_instance.create_api_key(create_api_key_request=create_request)
    print("API Key created successfully:")
    print(f"API Key ID: {api_response.api_key_metadata.api_key_id}")
    print(f"Raw API Key: {api_response.raw_api_key}")
except ApiException as e:
    print(f"Exception when calling APIKeysApi->create_api_key: {e}")

from goodmem_client.api import SpacesApi
from goodmem_client.streaming import MemoryStreamClient
from goodmem_client.configuration import Configuration
from goodmem_client.api_client import ApiClient

## Configure client
configuration = Configuration()
configuration.host = "http://localhost:8080"
configuration.api_key = {"ApiKeyAuth": "your-api-key"}

api_client = ApiClient(configuration=configuration)

## Create streaming client
spaces_api = SpacesApi(api_client=api_client)
stream_client = MemoryStreamClient(api_client)

## Get space ID (first available space)
spaces = spaces_api.list_spaces()
space_id = spaces.spaces[0].space_id if spaces.spaces else None

## Stream memory retrieval using advanced POST endpoint with custom post-processor
for event in stream_client.retrieve_memory_stream(
    message="your search query",
    space_ids=[space_id] if space_id else None,
    requested_size=5,
    fetch_memory=True,
    fetch_memory_content=False,
    format="ndjson",
    post_processor_name="com.goodmem.retrieval.postprocess.ChatPostProcessorFactory",
    post_processor_config={
        "llm_id": "your-llm-uuid",
        "reranker_id": "your-reranker-uuid",
        "relevance_threshold": 0.5,
        "max_results": 10
    }
):
    if event.abstract_reply:
        print(f"Abstract: {event.abstract_reply.text}")
    elif event.retrieved_item and event.retrieved_item.memory:
        memory = event.retrieved_item.memory
        print(f"Memory: {memory.get('memoryId')}")
        if 'metadata' in memory:
            print(f"Metadata: {memory['metadata']}")

## Alternative: Use convenience method for ChatPostProcessor with simple parameters
for event in stream_client.retrieve_memory_stream_chat(
    message="your search query",
    space_ids=[space_id] if space_id else None,
    requested_size=5,
    fetch_memory=True,
    format="ndjson",
    pp_llm_id="your-llm-uuid",
    pp_reranker_id="your-reranker-uuid",
    pp_relevance_threshold=0.5,
    pp_max_results=10
):
    if event.abstract_reply:
        print(f"Abstract: {event.abstract_reply.text}")

from goodmem_client.streaming import MemoryStreamClient
from goodmem_client.models.space_key import SpaceKey
from goodmem_client.configuration import Configuration
from goodmem_client.api_client import ApiClient

## Configure client
configuration = Configuration()
configuration.host = "http://localhost:8080"
configuration.api_key = {"ApiKeyAuth": "your-api-key"}

api_client = ApiClient(configuration=configuration)
stream_client = MemoryStreamClient(api_client)

## Example 1: Filter by category
space_key = SpaceKey(
    space_id="your-space-id",
    filter="CAST(val('$.category') AS TEXT) = 'technology'"
)

for event in stream_client.retrieve_memory_stream(
    message="artificial intelligence",
    space_keys=[space_key],
    requested_size=5
):
    if event.retrieved_item and event.retrieved_item.chunk:
        chunk = event.retrieved_item.chunk
        print(f"Chunk: {chunk.chunk.get('chunkText', '')[:100]}...")

## Example 2: Filter by time range
space_key = SpaceKey(
    space_id="your-space-id",
    filter="CAST(val('$.created_at') AS TEXT) >= '2025-01-01T00:00:00Z'"
)

for event in stream_client.retrieve_memory_stream(
    message="recent developments",
    space_keys=[space_key],
    requested_size=5
):
    if event.retrieved_item and event.retrieved_item.chunk:
        chunk = event.retrieved_item.chunk
        print(f"Relevance: {chunk.relevance_score:.4f}")
        print(f"Text: {chunk.chunk.get('chunkText', '')[:80]}...")

## Example 3: Complex filter with multiple conditions
space_key = SpaceKey(
    space_id="your-space-id",
    filter="""
        CAST(val('$.priority') AS TEXT) = 'high'
        AND CAST(val('$.status') AS TEXT) = 'active'
        AND CAST(val('$.score') AS INTEGER) >= 80
    """
)

for event in stream_client.retrieve_memory_stream(
    message="important tasks",
    space_keys=[space_key],
    requested_size=10
):
    if event.retrieved_item and event.retrieved_item.chunk:
        print(f"Found matching chunk")

## Example 4: Different filters for different spaces
space_keys = [
    SpaceKey(
        space_id="space-1-id",
        filter="CAST(val('$.category') AS TEXT) = 'research'"
    ),
    SpaceKey(
        space_id="space-2-id",
        filter="CAST(val('$.category') AS TEXT) = 'documentation'"
    )
]

for event in stream_client.retrieve_memory_stream(
    message="machine learning concepts",
    space_keys=space_keys,
    requested_size=10
):
    if event.memory_definition:
        space_id = event.memory_definition.get('spaceId')
        metadata = event.memory_definition.get('metadata', {})
        print(f"Memory from space: {space_id}, category: {metadata.get('category')}")

## Example 5: Mix filtered and unfiltered spaces
space_keys = [
    SpaceKey(
        space_id="space-1-id",
        filter="CAST(val('$.source') AS TEXT) = 'trusted_source'"
    ),
    SpaceKey(
        space_id="space-2-id"  # No filter - all memories from this space
    )
]

for event in stream_client.retrieve_memory_stream(
    message="information query",
    space_keys=space_keys,
    requested_size=10
):
    if event.retrieved_item and event.retrieved_item.chunk:
        print(f"Relevance: {event.retrieved_item.chunk.relevance_score:.4f}")