feat(openai-sdk-python): expand OpenAI tools to 7-tool parity

Add document_list, document_add, document_delete, get_profile, and
memory_forget alongside search_memories and add_memory. Includes
forget_memory HTTP helper and expanded tests.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Dhravya Shah 2026-08-07 19:39:24 -07:00
parent 03849ce94f
commit 1160c46ccc
6 changed files with 782 additions and 150 deletions

View file

@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project]
name = "supermemory-openai-sdk"
version = "1.0.4"
version = "1.0.6"
description = "Memory tools for OpenAI function calling with supermemory"
readme = "README.md"
license = "MIT"
@ -15,7 +15,6 @@ classifiers = [
"Intended Audience :: Developers",
"License :: OSI Approved :: MIT License",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.8",
"Programming Language :: Python :: 3.9",
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11",
@ -26,7 +25,7 @@ classifiers = [
requires-python = ">=3.9"
dependencies = [
"openai>=1.102.0",
"supermemory>=3.50.0",
"supermemory>=3.16.0",
"typing-extensions>=4.0.0",
"requests>=2.25.0",
]
@ -62,7 +61,7 @@ multi_line_output = 3
line_length = 88
[tool.mypy]
python_version = "3.8"
python_version = "3.9"
warn_return_any = true
warn_unused_configs = true
disallow_untyped_defs = true

View file

@ -6,14 +6,29 @@ from .tools import (
MemoryObject,
MemorySearchResult,
MemoryAddResult,
ProfileResult,
DocumentListResult,
DocumentDeleteResult,
DocumentAddResult,
MemoryForgetResult,
SearchMemoriesTool,
AddMemoryTool,
GetProfileTool,
DocumentListTool,
DocumentDeleteTool,
DocumentAddTool,
MemoryForgetTool,
MEMORY_TOOL_SCHEMAS,
create_supermemory_tools,
get_memory_tool_definitions,
execute_memory_tool_calls,
create_search_memories_tool,
create_add_memory_tool,
create_get_profile_tool,
create_document_list_tool,
create_document_delete_tool,
create_document_add_tool,
create_memory_forget_tool,
)
from .middleware import (
@ -48,14 +63,29 @@ __all__ = [
"MemoryObject",
"MemorySearchResult",
"MemoryAddResult",
"ProfileResult",
"DocumentListResult",
"DocumentDeleteResult",
"DocumentAddResult",
"MemoryForgetResult",
"SearchMemoriesTool",
"AddMemoryTool",
"GetProfileTool",
"DocumentListTool",
"DocumentDeleteTool",
"DocumentAddTool",
"MemoryForgetTool",
"MEMORY_TOOL_SCHEMAS",
"create_supermemory_tools",
"get_memory_tool_definitions",
"execute_memory_tool_calls",
"create_search_memories_tool",
"create_add_memory_tool",
"create_get_profile_tool",
"create_document_list_tool",
"create_document_delete_tool",
"create_document_add_tool",
"create_memory_forget_tool",
# Middleware
"with_supermemory",
"OpenAIMiddlewareOptions",

View file

@ -0,0 +1,59 @@
"""Forget memory via DELETE /v4/memories (not exposed on supermemory SDK v3)."""
from typing import Optional
DEFAULT_BASE_URL = "https://api.supermemory.ai"
async def forget_memory_request(
api_key: str,
container_tag: str,
memory_id: Optional[str] = None,
memory_content: Optional[str] = None,
reason: Optional[str] = None,
base_url: str = DEFAULT_BASE_URL,
) -> None:
"""Mark a memory as forgotten via the v4 memories endpoint."""
payload: dict[str, str] = {"containerTag": container_tag}
if memory_id:
payload["id"] = memory_id
if memory_content:
payload["content"] = memory_content
if reason:
payload["reason"] = reason
try:
import aiohttp
async with aiohttp.ClientSession() as session:
async with session.delete(
f"{base_url}/v4/memories",
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {api_key}",
},
json=payload,
) as response:
if not response.ok:
error_text = await response.text()
raise RuntimeError(
f"Supermemory forget memory failed: {response.status} "
f"{response.reason}. {error_text}"
)
except ImportError:
import requests
response = requests.delete(
f"{base_url}/v4/memories",
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {api_key}",
},
json=payload,
timeout=30,
)
if not response.ok:
raise RuntimeError(
f"Supermemory forget memory failed: {response.status_code} "
f"{response.reason}. {response.text}"
)

View file

@ -1,15 +1,20 @@
"""Supermemory tools for OpenAI function calling."""
import json
from typing import Dict, List, Optional, TypedDict, Union
from typing import Any, Dict, List, Optional, TypedDict, Union
import supermemory
from openai.types.chat import (
ChatCompletionFunctionToolParam,
ChatCompletionMessageToolCall,
ChatCompletionToolMessageParam,
ChatCompletionToolParam,
)
from supermemory.types import (
AddResponse,
DocumentGetResponse,
SearchMemoriesResponse,
)
from supermemory.types import AddResponse, SearchMemoriesResponse
from supermemory.types.search_memories_response import Result
from .exceptions import (
@ -17,6 +22,109 @@ from .exceptions import (
SupermemoryMemoryOperationError,
SupermemoryNetworkError,
)
from .forget_memory import DEFAULT_BASE_URL, forget_memory_request
TOOL_DESCRIPTIONS = {
"search_memories": (
"Search (recall) stored memories for facts, preferences, history, and context about the user "
"or any topic. Use proactively before answering whenever memory could help — do not wait for "
"the user to explicitly ask you to search or recall. Search when the question touches personal "
"context, past conversations, preferences, projects, people, plans, or anything you may have "
"learned before. Results include memory/chunk IDs — use those IDs with memory_forget to remove "
"a specific learned fact."
),
"add_memory": (
"Add (remember) memories/details/information about the user or other facts or entities. "
"Run when explicitly asked or when the user mentions any information generalizable beyond "
"the context of the current conversation."
),
"get_profile": (
"Get user profile containing static memories (permanent facts) and dynamic memories "
"(recent context). Optionally include search results by providing a query. "
"Profile and search result entries may include memory IDs useful for memory_forget."
),
"document_list": (
"List stored source documents (conversations, URLs, files, pasted text) with pagination. "
"Returns document IDs for document_delete — not memory IDs for memory_forget. "
"Use to browse raw stored content before permanently removing a source."
),
"document_delete": (
"Permanently delete a stored document and ALL memories extracted from it (hard delete). "
"Use document IDs from document_list. Use when the user wants to remove an entire "
"conversation, file, URL, or other source — not when correcting a single learned fact "
"(use memory_forget for that)."
),
"document_add": (
"Store a source document for asynchronous processing and automatic memory extraction. "
"Use when the user gives you raw content to ingest — a pasted text blob, conversation transcript, "
"chat history, notes, URL, article link, or other substantial text — rather than a single atomic "
"fact (use add_memory for one short generalizable sentence). The document is queued immediately; "
"Supermemory post-processes it in the background (chunking, embedding, indexing) and extracts profile "
"memories automatically — you do not need to call add_memory for facts buried inside the document. "
"Good for saving full conversations, long-form notes, knowledge-base articles, meeting transcripts, "
"or any large body of text the user wants remembered beyond this chat turn. Processing may take a "
"moment; extracted memories appear in profile/search after indexing completes."
),
"memory_forget": (
"Soft-delete a single extracted profile memory (a learned fact) so it no longer appears in "
"profile or search. Does NOT delete source documents. Provide memory_id (preferred — from "
"search_memories or get_profile) OR memory_content for an exact text match. Use when the "
"user retracts or corrects a specific fact (e.g. 'forget I like tea', 'that's wrong'). "
"To remove an entire conversation or file, use document_delete instead."
),
}
PARAMETER_DESCRIPTIONS = {
"information_to_get": (
"What to look up in memory — keywords from the user's message, topic, entity names, or "
"question phrasing. Search even when the user did not explicitly ask you to recall."
),
"include_full_docs": (
"Whether to include the full document content in the response. "
"Defaults to true for better AI context."
),
"limit": "Maximum number of results to return",
"memory": (
"The text content of the memory to add. This should be a single sentence or a short paragraph."
),
"container_tag": "Tag to filter/scope the operation (e.g., user ID, project ID)",
"query": "Optional search query to include relevant search results",
"page": "Page number to fetch, 1-based (default: 1)",
"document_id": (
"Document ID from document_list — permanently deletes the source document and all "
"extracted memories. Not a profile memory ID."
),
"content": (
"Document body to store — plain text, a conversation transcript, a long pasted blob, or a URL "
"to a webpage/PDF/image/video. Content is queued and memories are extracted automatically after "
"background processing; do not split into add_memory calls."
),
"title": "Optional title for the document",
"description": "Optional description for the document",
"memory_id": (
"Profile memory ID from search_memories or get_profile — soft-deletes one learned fact. "
"Not a document ID."
),
"memory_content": (
"Exact text of the profile memory to forget (alternative to memory_id). Must match "
"precisely; if unsure, search first and use memory_id."
),
"reason": "Optional reason recorded when forgetting (e.g. outdated, user correction)",
}
DEFAULT_LIMIT = 10
DEFAULT_CHUNK_THRESHOLD = 0.6
DEFAULT_INCLUDE_FULL_DOCS = True
ALL_TOOL_NAMES = (
"search_memories",
"add_memory",
"get_profile",
"document_list",
"document_delete",
"document_add",
"memory_forget",
)
class SupermemoryToolsConfig(TypedDict, total=False):
@ -31,7 +139,7 @@ class SupermemoryToolsConfig(TypedDict, total=False):
# Type aliases using inferred types from supermemory package
MemoryObject = AddResponse
MemoryObject = Union[DocumentGetResponse, AddResponse]
class MemorySearchResult(TypedDict, total=False):
@ -51,32 +159,69 @@ class MemoryAddResult(TypedDict, total=False):
error: Optional[str]
class ProfileResult(TypedDict, total=False):
"""Result type for profile operations."""
success: bool
profile: Optional[Dict[str, Any]]
search_results: Optional[Any]
error: Optional[str]
class DocumentListResult(TypedDict, total=False):
"""Result type for document list operations."""
success: bool
documents: Optional[List[Any]]
pagination: Optional[Any]
error: Optional[str]
class DocumentDeleteResult(TypedDict, total=False):
"""Result type for document delete operations."""
success: bool
message: Optional[str]
error: Optional[str]
class DocumentAddResult(TypedDict, total=False):
"""Result type for document add operations."""
success: bool
document: Optional[Dict[str, object]]
error: Optional[str]
class MemoryForgetResult(TypedDict, total=False):
"""Result type for memory forget operations."""
success: bool
message: Optional[str]
error: Optional[str]
# Function schemas for OpenAI function calling
MEMORY_TOOL_SCHEMAS: Dict[str, ChatCompletionFunctionToolParam] = {
"search_memories": {
"name": "search_memories",
"description": (
"Search (recall) memories/details/information about the user or other facts or entities. Run when explicitly asked or when context about user's past choices would be helpful."
),
"description": TOOL_DESCRIPTIONS["search_memories"],
"parameters": {
"type": "object",
"properties": {
"information_to_get": {
"type": "string",
"description": "Terms to search for in the user's memories",
"description": PARAMETER_DESCRIPTIONS["information_to_get"],
},
"include_full_docs": {
"type": "boolean",
"description": (
"Whether to include the full document content in the response. "
"Defaults to true for better AI context."
),
"default": True,
"description": PARAMETER_DESCRIPTIONS["include_full_docs"],
"default": DEFAULT_INCLUDE_FULL_DOCS,
},
"limit": {
"type": "number",
"description": "Maximum number of results to return",
"default": 10,
"description": PARAMETER_DESCRIPTIONS["limit"],
"default": DEFAULT_LIMIT,
},
},
"required": ["information_to_get"],
@ -84,26 +229,144 @@ MEMORY_TOOL_SCHEMAS: Dict[str, ChatCompletionFunctionToolParam] = {
},
"add_memory": {
"name": "add_memory",
"description": (
"Add (remember) memories/details/information about the user or other facts or entities. Run when explicitly asked or when the user mentions any information generalizable beyond the context of the current conversation."
),
"description": TOOL_DESCRIPTIONS["add_memory"],
"parameters": {
"type": "object",
"properties": {
"memory": {
"type": "string",
"description": (
"The text content of the memory to add. This should be a "
"single sentence or a short paragraph."
),
"description": PARAMETER_DESCRIPTIONS["memory"],
},
},
"required": ["memory"],
},
},
"get_profile": {
"name": "get_profile",
"description": TOOL_DESCRIPTIONS["get_profile"],
"parameters": {
"type": "object",
"properties": {
"container_tag": {
"type": "string",
"description": PARAMETER_DESCRIPTIONS["container_tag"],
},
"query": {
"type": "string",
"description": PARAMETER_DESCRIPTIONS["query"],
},
},
"required": [],
},
},
"document_list": {
"name": "document_list",
"description": TOOL_DESCRIPTIONS["document_list"],
"parameters": {
"type": "object",
"properties": {
"container_tag": {
"type": "string",
"description": PARAMETER_DESCRIPTIONS["container_tag"],
},
"limit": {
"type": "number",
"description": PARAMETER_DESCRIPTIONS["limit"],
"default": DEFAULT_LIMIT,
},
"page": {
"type": "number",
"description": PARAMETER_DESCRIPTIONS["page"],
},
},
"required": [],
},
},
"document_delete": {
"name": "document_delete",
"description": TOOL_DESCRIPTIONS["document_delete"],
"parameters": {
"type": "object",
"properties": {
"document_id": {
"type": "string",
"description": PARAMETER_DESCRIPTIONS["document_id"],
},
},
"required": ["document_id"],
},
},
"document_add": {
"name": "document_add",
"description": TOOL_DESCRIPTIONS["document_add"],
"parameters": {
"type": "object",
"properties": {
"content": {
"type": "string",
"description": PARAMETER_DESCRIPTIONS["content"],
},
"title": {
"type": "string",
"description": PARAMETER_DESCRIPTIONS["title"],
},
"description": {
"type": "string",
"description": PARAMETER_DESCRIPTIONS["description"],
},
},
"required": ["content"],
},
},
"memory_forget": {
"name": "memory_forget",
"description": TOOL_DESCRIPTIONS["memory_forget"],
"parameters": {
"type": "object",
"properties": {
"container_tag": {
"type": "string",
"description": PARAMETER_DESCRIPTIONS["container_tag"],
},
"memory_id": {
"type": "string",
"description": PARAMETER_DESCRIPTIONS["memory_id"],
},
"memory_content": {
"type": "string",
"description": PARAMETER_DESCRIPTIONS["memory_content"],
},
"reason": {
"type": "string",
"description": PARAMETER_DESCRIPTIONS["reason"],
},
},
"required": [],
},
},
}
def _resolve_container_tags(config: SupermemoryToolsConfig) -> List[str]:
if config.get("project_id") is not None and config.get("container_tags") is not None:
raise SupermemoryConfigurationError(
"Supermemory tools config accepts either project_id or container_tags, not both."
)
if config.get("project_id"):
return [f"sm_project_{config['project_id']}"]
if config.get("container_tags"):
return config["container_tags"]
return ["sm_project_default"]
def _tool_definition(name: str) -> ChatCompletionToolParam:
return {"type": "function", "function": MEMORY_TOOL_SCHEMAS[name]}
def _all_tool_definitions() -> List[ChatCompletionFunctionToolParam]:
return [{"type": "function", "function": MEMORY_TOOL_SCHEMAS[name]} for name in ALL_TOOL_NAMES]
class SupermemoryTools:
"""Create memory tool handlers for OpenAI function calling."""
@ -115,79 +378,62 @@ class SupermemoryTools:
config: Optional configuration
"""
config = config or {}
self.api_key = api_key
self.base_url = config.get("base_url") or DEFAULT_BASE_URL
# Initialize Supermemory client
client_kwargs = {"api_key": api_key}
if config.get("base_url"):
client_kwargs["base_url"] = config["base_url"]
self.client = supermemory.AsyncSupermemory(**client_kwargs)
self.container_tags = _resolve_container_tags(config)
# Set container tags
if config.get("project_id"):
self.container_tags = [f"sm_project_{config['project_id']}"]
elif config.get("container_tags"):
self.container_tags = config["container_tags"]
else:
self.container_tags = ["sm_project_default"]
def _primary_container_tag(self, container_tag: Optional[str] = None) -> str:
return container_tag or self.container_tags[0]
def get_tool_definitions(self) -> List[ChatCompletionFunctionToolParam]:
"""Get OpenAI function definitions for all memory tools.
Returns:
List of ChatCompletionToolParam definitions
"""
return [
{"type": "function", "function": MEMORY_TOOL_SCHEMAS["search_memories"]},
{"type": "function", "function": MEMORY_TOOL_SCHEMAS["add_memory"]},
]
"""Get OpenAI function definitions for all memory tools."""
return _all_tool_definitions()
async def execute_tool_call(self, tool_call: ChatCompletionMessageToolCall) -> str:
"""Execute a tool call based on the function name and arguments.
Args:
tool_call: The tool call from OpenAI
Returns:
JSON string result
"""
"""Execute a tool call based on the function name and arguments."""
function_name = tool_call.function.name
args = json.loads(tool_call.function.arguments)
if function_name == "search_memories":
result = await self.search_memories(**args)
elif function_name == "add_memory":
result = await self.add_memory(**args)
else:
result = {
handlers = {
"search_memories": self.search_memories,
"add_memory": self.add_memory,
"get_profile": self.get_profile,
"document_list": self.document_list,
"document_delete": self.document_delete,
"document_add": self.document_add,
"memory_forget": self.memory_forget,
}
handler = handlers.get(function_name)
if handler is None:
result: Dict[str, object] = {
"success": False,
"error": f"Unknown function: {function_name}",
}
else:
result = await handler(**args)
return json.dumps(result)
async def search_memories(
self,
information_to_get: str,
include_full_docs: bool = True,
limit: int = 10,
include_full_docs: bool = DEFAULT_INCLUDE_FULL_DOCS,
limit: int = DEFAULT_LIMIT,
) -> MemorySearchResult:
"""Search memories.
Args:
information_to_get: Terms to search for
include_full_docs: Whether to include full document content
limit: Maximum number of results
Returns:
MemorySearchResult
"""
"""Search memories."""
try:
response: SearchMemoriesResponse = await self.client.search.memories(
q=information_to_get,
container_tags=self.container_tags,
limit=limit,
threshold=0.6,
threshold=DEFAULT_CHUNK_THRESHOLD,
search_mode="hybrid",
)
@ -209,14 +455,7 @@ class SupermemoryTools:
)
async def add_memory(self, memory: str) -> MemoryAddResult:
"""Add a memory.
Args:
memory: The memory content to add
Returns:
MemoryAddResult
"""
"""Add a memory."""
try:
response: AddResponse = await self.client.add(
content=memory,
@ -238,32 +477,181 @@ class SupermemoryTools:
error=f"Memory add failed: {error}",
)
async def get_profile(
self,
container_tag: Optional[str] = None,
query: Optional[str] = None,
) -> ProfileResult:
"""Get user profile with optional query-scoped search results."""
try:
kwargs: Dict[str, Any] = {
"container_tag": self._primary_container_tag(container_tag),
}
if query:
kwargs["q"] = query
response = await self.client.profile(**kwargs)
profile = response.profile if hasattr(response, "profile") else None
search_results = (
response.search_results if hasattr(response, "search_results") else None
)
return ProfileResult(
success=True,
profile=profile if isinstance(profile, dict) else None,
search_results=search_results,
)
except (OSError, ConnectionError) as network_error:
return ProfileResult(
success=False,
error=f"Network error: {network_error}",
)
except Exception as error:
return ProfileResult(
success=False,
error=f"Profile fetch failed: {error}",
)
async def document_list(
self,
container_tag: Optional[str] = None,
limit: Optional[int] = None,
page: Optional[int] = None,
) -> DocumentListResult:
"""List stored documents."""
try:
kwargs: Dict[str, Any] = {
"container_tags": [self._primary_container_tag(container_tag)],
"limit": limit or DEFAULT_LIMIT,
}
if page is not None:
kwargs["page"] = page
response = await self.client.documents.list(**kwargs)
documents = response.memories if hasattr(response, "memories") else []
pagination = response.pagination if hasattr(response, "pagination") else None
return DocumentListResult(
success=True,
documents=documents,
pagination=pagination,
)
except (OSError, ConnectionError) as network_error:
return DocumentListResult(
success=False,
error=f"Network error: {network_error}",
)
except Exception as error:
return DocumentListResult(
success=False,
error=f"Document list failed: {error}",
)
async def document_delete(self, document_id: str) -> DocumentDeleteResult:
"""Delete a document by ID."""
try:
await self.client.documents.delete(document_id)
return DocumentDeleteResult(
success=True,
message=f"Document {document_id} deleted successfully",
)
except (OSError, ConnectionError) as network_error:
return DocumentDeleteResult(
success=False,
error=f"Network error: {network_error}",
)
except Exception as error:
return DocumentDeleteResult(
success=False,
error=f"Document delete failed: {error}",
)
async def document_add(
self,
content: str,
title: Optional[str] = None,
description: Optional[str] = None,
) -> DocumentAddResult:
"""Add a document for processing."""
try:
metadata: Dict[str, str] = {}
if title:
metadata["title"] = title
if description:
metadata["description"] = description
kwargs: Dict[str, Any] = {
"content": content,
"container_tags": self.container_tags,
}
if metadata:
kwargs["metadata"] = metadata
response = await self.client.documents.add(**kwargs)
return DocumentAddResult(
success=True,
document=response.model_dump(),
)
except (OSError, ConnectionError) as network_error:
return DocumentAddResult(
success=False,
error=f"Network error: {network_error}",
)
except Exception as error:
return DocumentAddResult(
success=False,
error=f"Document add failed: {error}",
)
async def memory_forget(
self,
container_tag: Optional[str] = None,
memory_id: Optional[str] = None,
memory_content: Optional[str] = None,
reason: Optional[str] = None,
) -> MemoryForgetResult:
"""Forget a memory by ID or content match."""
if not memory_id and not memory_content:
return MemoryForgetResult(
success=False,
error="Either memory_id or memory_content must be provided",
)
try:
await forget_memory_request(
api_key=self.api_key,
container_tag=self._primary_container_tag(container_tag),
memory_id=memory_id,
memory_content=memory_content,
reason=reason,
base_url=self.base_url,
)
return MemoryForgetResult(
success=True,
message="Memory forgotten successfully",
)
except (OSError, ConnectionError) as network_error:
return MemoryForgetResult(
success=False,
error=f"Network error: {network_error}",
)
except Exception as error:
return MemoryForgetResult(
success=False,
error=f"Memory forget failed: {error}",
)
def create_supermemory_tools(
api_key: str, config: Optional[SupermemoryToolsConfig] = None
) -> SupermemoryTools:
"""Helper function to create SupermemoryTools instance.
Args:
api_key: Supermemory API key
config: Optional configuration
Returns:
SupermemoryTools instance
"""
"""Helper function to create SupermemoryTools instance."""
return SupermemoryTools(api_key, config)
def get_memory_tool_definitions() -> List[ChatCompletionFunctionToolParam]:
"""Get OpenAI function definitions for memory tools.
Returns:
List of ChatCompletionToolParam definitions
"""
return [
{"type": "function", "function": MEMORY_TOOL_SCHEMAS["search_memories"]},
{"type": "function", "function": MEMORY_TOOL_SCHEMAS["add_memory"]},
]
"""Get OpenAI function definitions for memory tools."""
return _all_tool_definitions()
async def execute_memory_tool_calls(
@ -271,16 +659,7 @@ async def execute_memory_tool_calls(
tool_calls: List[ChatCompletionMessageToolCall],
config: Optional[SupermemoryToolsConfig] = None,
) -> List[ChatCompletionToolMessageParam]:
"""Execute tool calls from OpenAI function calling.
Args:
api_key: Supermemory API key
tool_calls: List of tool calls from OpenAI
config: Optional configuration
Returns:
List of tool message parameters
"""
"""Execute tool calls from OpenAI function calling."""
tools = SupermemoryTools(api_key, config)
async def execute_single_call(
@ -293,7 +672,6 @@ async def execute_memory_tool_calls(
content=result,
)
# Execute all tool calls concurrently
import asyncio
results = await asyncio.gather(
@ -303,22 +681,18 @@ async def execute_memory_tool_calls(
return results
# Individual tool creators for more granular control
class SearchMemoriesTool:
"""Individual search memories tool."""
def __init__(self, api_key: str, config: Optional[SupermemoryToolsConfig] = None):
self.tools = SupermemoryTools(api_key, config)
self.definition: ChatCompletionToolParam = {
"type": "function",
"function": MEMORY_TOOL_SCHEMAS["search_memories"],
}
self.definition: ChatCompletionToolParam = _tool_definition("search_memories")
async def execute(
self,
information_to_get: str,
include_full_docs: bool = True,
limit: int = 10,
include_full_docs: bool = DEFAULT_INCLUDE_FULL_DOCS,
limit: int = DEFAULT_LIMIT,
) -> MemorySearchResult:
"""Execute search memories."""
return await self.tools.search_memories(
@ -333,41 +707,150 @@ class AddMemoryTool:
def __init__(self, api_key: str, config: Optional[SupermemoryToolsConfig] = None):
self.tools = SupermemoryTools(api_key, config)
self.definition: ChatCompletionToolParam = {
"type": "function",
"function": MEMORY_TOOL_SCHEMAS["add_memory"],
}
self.definition: ChatCompletionToolParam = _tool_definition("add_memory")
async def execute(self, memory: str) -> MemoryAddResult:
"""Execute add memory."""
return await self.tools.add_memory(memory=memory)
class GetProfileTool:
"""Individual get profile tool."""
def __init__(self, api_key: str, config: Optional[SupermemoryToolsConfig] = None):
self.tools = SupermemoryTools(api_key, config)
self.definition: ChatCompletionToolParam = _tool_definition("get_profile")
async def execute(
self,
container_tag: Optional[str] = None,
query: Optional[str] = None,
) -> ProfileResult:
"""Execute get profile."""
return await self.tools.get_profile(container_tag=container_tag, query=query)
class DocumentListTool:
"""Individual document list tool."""
def __init__(self, api_key: str, config: Optional[SupermemoryToolsConfig] = None):
self.tools = SupermemoryTools(api_key, config)
self.definition: ChatCompletionToolParam = _tool_definition("document_list")
async def execute(
self,
container_tag: Optional[str] = None,
limit: Optional[int] = None,
page: Optional[int] = None,
) -> DocumentListResult:
"""Execute document list."""
return await self.tools.document_list(
container_tag=container_tag,
limit=limit,
page=page,
)
class DocumentDeleteTool:
"""Individual document delete tool."""
def __init__(self, api_key: str, config: Optional[SupermemoryToolsConfig] = None):
self.tools = SupermemoryTools(api_key, config)
self.definition: ChatCompletionToolParam = _tool_definition("document_delete")
async def execute(self, document_id: str) -> DocumentDeleteResult:
"""Execute document delete."""
return await self.tools.document_delete(document_id=document_id)
class DocumentAddTool:
"""Individual document add tool."""
def __init__(self, api_key: str, config: Optional[SupermemoryToolsConfig] = None):
self.tools = SupermemoryTools(api_key, config)
self.definition: ChatCompletionToolParam = _tool_definition("document_add")
async def execute(
self,
content: str,
title: Optional[str] = None,
description: Optional[str] = None,
) -> DocumentAddResult:
"""Execute document add."""
return await self.tools.document_add(
content=content,
title=title,
description=description,
)
class MemoryForgetTool:
"""Individual memory forget tool."""
def __init__(self, api_key: str, config: Optional[SupermemoryToolsConfig] = None):
self.tools = SupermemoryTools(api_key, config)
self.definition: ChatCompletionToolParam = _tool_definition("memory_forget")
async def execute(
self,
container_tag: Optional[str] = None,
memory_id: Optional[str] = None,
memory_content: Optional[str] = None,
reason: Optional[str] = None,
) -> MemoryForgetResult:
"""Execute memory forget."""
return await self.tools.memory_forget(
container_tag=container_tag,
memory_id=memory_id,
memory_content=memory_content,
reason=reason,
)
def create_search_memories_tool(
api_key: str, config: Optional[SupermemoryToolsConfig] = None
) -> SearchMemoriesTool:
"""Create individual search memories tool.
Args:
api_key: Supermemory API key
config: Optional configuration
Returns:
SearchMemoriesTool instance
"""
"""Create individual search memories tool."""
return SearchMemoriesTool(api_key, config)
def create_add_memory_tool(
api_key: str, config: Optional[SupermemoryToolsConfig] = None
) -> AddMemoryTool:
"""Create individual add memory tool.
Args:
api_key: Supermemory API key
config: Optional configuration
Returns:
AddMemoryTool instance
"""
"""Create individual add memory tool."""
return AddMemoryTool(api_key, config)
def create_get_profile_tool(
api_key: str, config: Optional[SupermemoryToolsConfig] = None
) -> GetProfileTool:
"""Create individual get profile tool."""
return GetProfileTool(api_key, config)
def create_document_list_tool(
api_key: str, config: Optional[SupermemoryToolsConfig] = None
) -> DocumentListTool:
"""Create individual document list tool."""
return DocumentListTool(api_key, config)
def create_document_delete_tool(
api_key: str, config: Optional[SupermemoryToolsConfig] = None
) -> DocumentDeleteTool:
"""Create individual document delete tool."""
return DocumentDeleteTool(api_key, config)
def create_document_add_tool(
api_key: str, config: Optional[SupermemoryToolsConfig] = None
) -> DocumentAddTool:
"""Create individual document add tool."""
return DocumentAddTool(api_key, config)
def create_memory_forget_tool(
api_key: str, config: Optional[SupermemoryToolsConfig] = None
) -> MemoryForgetTool:
"""Create individual memory forget tool."""
return MemoryForgetTool(api_key, config)

View file

@ -43,6 +43,8 @@ except ImportError:
# SupermemoryOpenAI,
# SupermemoryInfiniteChatConfigWithProviderName,
EXPECTED_TOOL_COUNT = 7
@pytest.fixture
def test_api_key() -> str:
@ -84,9 +86,7 @@ class TestToolInitialization:
assert tools is not None
assert tools.get_tool_definitions() is not None
assert (
len(tools.get_tool_definitions()) == 2
) # Currently has search_memories and add_memory
assert len(tools.get_tool_definitions()) == EXPECTED_TOOL_COUNT
def test_create_tools_with_helper(self, test_api_key: str):
"""Test creating tools with createSupermemoryTools helper."""
@ -113,9 +113,7 @@ class TestToolInitialization:
tools = SupermemoryTools(test_api_key, config)
assert tools is not None
assert (
len(tools.get_tool_definitions()) == 2
) # Currently has search_memories and add_memory
assert len(tools.get_tool_definitions()) == EXPECTED_TOOL_COUNT
def test_create_tools_with_project_id(self, test_api_key: str):
"""Test creating tools with projectId configuration."""
@ -125,9 +123,7 @@ class TestToolInitialization:
tools = SupermemoryTools(test_api_key, config)
assert tools is not None
assert (
len(tools.get_tool_definitions()) == 2
) # Currently has search_memories and add_memory
assert len(tools.get_tool_definitions()) == EXPECTED_TOOL_COUNT
def test_create_tools_with_custom_container_tags(self, test_api_key: str):
"""Test creating tools with custom container tags."""
@ -137,9 +133,7 @@ class TestToolInitialization:
tools = SupermemoryTools(test_api_key, config)
assert tools is not None
assert (
len(tools.get_tool_definitions()) == 2
) # Currently has search_memories and add_memory
assert len(tools.get_tool_definitions()) == EXPECTED_TOOL_COUNT
class TestToolDefinitions:
@ -150,7 +144,7 @@ class TestToolDefinitions:
definitions = get_memory_tool_definitions()
assert definitions is not None
assert len(definitions) == 2 # Currently has search_memories and add_memory
assert len(definitions) == EXPECTED_TOOL_COUNT
# Check searchMemories
search_tool = next(
@ -228,6 +222,73 @@ class TestMemoryOperationsUnit:
assert kwargs["limit"] == 3
assert kwargs["search_mode"] == "hybrid"
@pytest.mark.asyncio
async def test_get_profile_uses_client_profile(self):
"""get_profile must call client.profile with container tag and optional query."""
from types import SimpleNamespace
from unittest.mock import AsyncMock
tools = SupermemoryTools("test-key", {"container_tags": ["unit-tag"]})
tools.client.profile = AsyncMock(
return_value=SimpleNamespace(
profile={"static": ["likes tea"], "dynamic": []},
search_results={"results": []},
)
)
result = await tools.get_profile(query="tea")
assert result["success"] is True
tools.client.profile.assert_awaited_once_with(
container_tag="unit-tag",
q="tea",
)
@pytest.mark.asyncio
async def test_document_list_uses_client_documents_list(self):
"""document_list must call client.documents.list with container tag."""
from types import SimpleNamespace
from unittest.mock import AsyncMock
tools = SupermemoryTools("test-key", {"container_tags": ["unit-tag"]})
tools.client.documents.list = AsyncMock(
return_value=SimpleNamespace(
memories=[{"id": "doc_1"}],
pagination={"page": 1},
)
)
result = await tools.document_list(limit=5, page=2)
assert result["success"] is True
tools.client.documents.list.assert_awaited_once_with(
container_tags=["unit-tag"],
limit=5,
page=2,
)
@pytest.mark.asyncio
async def test_memory_forget_requires_id_or_content(self):
"""memory_forget must reject calls without memory_id or memory_content."""
tools = SupermemoryTools("test-key", {"container_tags": ["unit-tag"]})
result = await tools.memory_forget()
assert result["success"] is False
assert "memory_id or memory_content" in result["error"]
def test_rejects_project_id_and_container_tags(self):
"""Config must reject both project_id and container_tags."""
from supermemory_openai.exceptions import SupermemoryConfigurationError
with pytest.raises(SupermemoryConfigurationError):
SupermemoryTools(
"test-key",
{
"project_id": "abc",
"container_tags": ["tag-a"],
},
)
class TestMemoryOperations:
"""Test memory operations."""

View file

@ -1372,7 +1372,7 @@ wheels = [
[[package]]
name = "supermemory-openai-sdk"
version = "1.0.4"
version = "1.0.6"
source = { editable = "." }
dependencies = [
{ name = "openai" },
@ -1402,7 +1402,7 @@ requires-dist = [
{ name = "aiohttp", marker = "extra == 'async'", specifier = ">=3.8.0" },
{ name = "openai", specifier = ">=1.102.0" },
{ name = "requests", specifier = ">=2.25.0" },
{ name = "supermemory", specifier = ">=3.50.0" },
{ name = "supermemory", specifier = ">=3.16.0" },
{ name = "typing-extensions", specifier = ">=4.0.0" },
]
provides-extras = ["async"]