scriptoriumadmin/supermemory
1
0
Fork 0
mirror of https://github.com/supermemoryai/supermemory.git synced 2026-09-05 08:06:19 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 6

feat(openai-sdk-python): 7-tool parity (#1430)

Browse source 
## Summary
- Expand `SupermemoryTools` from 2 tools to 7 (matches `@supermemory/tools`)
- Add `memory_forget` via shared HTTP helper
- Add document list/add/delete and get_profile tool surfaces
- Expand tests for new tools and execution paths

Stacked on #1429

## Test plan
- [x] `uv run pytest tests/test_tools.py::TestMemoryOperationsUnit`

Made with [Cursor](https://cursor.com)
This commit is contained in:
Dhravya 2026-09-01 04:34:18 +00:00
parent c5b7e7d4fc
commit 348483d5e8
No known key found for this signature in database
GPG key ID: 135A27003CF4F6CB
6 changed files with 807 additions and 160 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

46
packages/openai-sdk-python/README.md
View file

@ -229,6 +229,17 @@ openai_with_memory = with_supermemory(
## Manual Memory Tools
`SupermemoryTools` exposes seven OpenAI function-calling tools:
- `search_memories` and `add_memory`
- `get_profile`
- `document_list`, `document_add`, and `document_delete`
- `memory_forget`
The configured `project_id` or `container_tags` define the trusted scope. The
primary tag is used for profile, list, search, and forget operations, and the
model cannot select a different tag.
### SupermemoryTools Class
```python
@ -253,9 +264,21 @@ result = await tools.add_memory(
memory="User prefers tea over coffee"
)
# Fetch specific memory
result = await tools.fetch_memory(
memory_id="memory-id-here"
# Get the configured user's profile
result = await tools.get_profile(query="favorite drinks")
# List, add, or delete source documents
documents = await tools.document_list(limit=10, page=1)
document = await tools.document_add(
content="Meeting notes...",
title="Weekly meeting"
)
deleted = await tools.document_delete(document_id="document-id-here")
# Soft-forget one extracted memory
forgotten = await tools.memory_forget(
memory_id="memory-entry-id-here",
reason="outdated"
)
```
@ -269,12 +292,20 @@ It is no longer exposed in the OpenAI tool schema.
from supermemory_openai import (
create_search_memories_tool,
create_add_memory_tool,
create_fetch_memory_tool
create_get_profile_tool,
create_document_list_tool,
create_document_delete_tool,
create_document_add_tool,
create_memory_forget_tool,
)
search_tool = create_search_memories_tool("your-api-key")
add_tool = create_add_memory_tool("your-api-key")
fetch_tool = create_fetch_memory_tool("your-api-key")
profile_tool = create_get_profile_tool("your-api-key")
list_tool = create_document_list_tool("your-api-key")
delete_tool = create_document_delete_tool("your-api-key")
document_add_tool = create_document_add_tool("your-api-key")
forget_tool = create_memory_forget_tool("your-api-key")
```
### Function Calling Integration
@ -346,6 +377,11 @@ SupermemoryTools(
- `get_tool_definitions()` - Get OpenAI function definitions
- `search_memories()` - Search user memories
- `add_memory()` - Add new memory
- `get_profile()` - Get the configured user's profile
- `document_list()` - List source document metadata
- `document_add()` - Queue a source document for processing
- `document_delete()` - Delete an in-scope source document
- `memory_forget()` - Soft-forget one extracted memory
- `execute_tool_call()` - Execute individual tool call
## Error Handling

2
packages/openai-sdk-python/pyproject.toml
View file

@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project]
name = "supermemory-openai-sdk"
version = "1.0.6"
version = "1.0.7"
description = "Memory tools for OpenAI function calling with supermemory"
readme = "README.md"
license = "MIT"

30
packages/openai-sdk-python/src/supermemory_openai/__init__.py
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",

794
packages/openai-sdk-python/src/supermemory_openai/tools.py
View file

@ -2,20 +2,111 @@
import json
import warnings
from typing import Dict, List, Optional, TypedDict
from typing import Any, Dict, List, Optional, TypedDict
import supermemory
from openai.types.chat import (
ChatCompletionFunctionToolParam,
ChatCompletionMessageToolCall,
ChatCompletionToolMessageParam,
ChatCompletionToolParam,
)
from openai.types.shared_params import FunctionDefinition
from supermemory.types import AddResponse, SearchMemoriesResponse
from .exceptions import (
SupermemoryConfigurationError,
SupermemoryMemoryOperationError,
SupermemoryNetworkError,
from .exceptions import SupermemoryConfigurationError
TOOL_DESCRIPTIONS = {
"search_memories": (
"Search stored memories and source chunks for relevant facts, preferences, "
"history, and context. Use proactively whenever prior context could help. "
"Hybrid results may contain either a memory or a source chunk; only an ID "
"from a result containing a memory can be passed to memory_forget."
),
"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 query-relevant search results. Static and dynamic "
"profile entries are text; only memory entries in search results have forgettable IDs."
),
"document_list": (
"List stored source documents (conversations, URLs, files, pasted text) with pagination. "
"Returns document metadata and summaries, including IDs for document_delete. "
"It does not return full source content or memory IDs."
),
"document_delete": (
"Permanently delete a stored source document and soft-forget memories extracted from it. "
"Use a document ID from document_list when the user wants to remove an entire source. "
"Deletion is refused for documents outside the configured scope, shared with another "
"scope, or still processing. Use memory_forget to remove one learned fact."
),
"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. This does not delete source documents. Provide a memory_id from a "
"search result containing a memory, or memory_content for an exact text match. Use "
"document_delete to remove an entire source."
),
}
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."
),
"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."
),
"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 and soft-forgets its "
"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": (
"Memory entry ID from a search_memories result containing a memory. Chunk and document "
"IDs are invalid."
),
"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
ALL_TOOL_NAMES = (
"search_memories",
"add_memory",
"get_profile",
"document_list",
"document_delete",
"document_add",
"memory_forget",
)
@ -23,8 +114,8 @@ class SupermemoryToolsConfig(TypedDict, total=False):
"""Configuration for Supermemory tools.
Only one of `project_id` or `container_tags` can be provided.
The first container tag is the primary v4 search scope; all configured tags
are applied when adding a memory.
The first container tag is used for single-space operations. All configured
tags are applied to additions and define the allowed document-delete scope.
"""
base_url: Optional[str]
@ -32,7 +123,7 @@ class SupermemoryToolsConfig(TypedDict, total=False):
project_id: Optional[str]
# Type aliases using inferred types from supermemory package
# Type alias retained for compatibility with earlier releases.
MemoryObject = AddResponse
@ -53,51 +144,234 @@ class MemoryAddResult(TypedDict, total=False):
error: Optional[str]
class ProfileResult(TypedDict, total=False):
"""Result type for profile operations."""
success: bool
profile: Optional[Dict[str, object]]
search_results: Optional[Dict[str, object]]
error: Optional[str]
class DocumentListResult(TypedDict, total=False):
"""Result type for document list operations."""
success: bool
documents: Optional[List[Dict[str, object]]]
pagination: Optional[Dict[str, object]]
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] = {
MEMORY_TOOL_SCHEMAS: Dict[str, FunctionDefinition] = {
"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"],
},
"limit": {
"type": "number",
"description": "Maximum number of results to return",
"default": 10,
"type": "integer",
"description": PARAMETER_DESCRIPTIONS["limit"],
"default": DEFAULT_LIMIT,
"minimum": 1,
"maximum": 100,
},
},
"required": ["information_to_get"],
"additionalProperties": False,
},
},
"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"],
"additionalProperties": False,
},
},
"get_profile": {
"name": "get_profile",
"description": TOOL_DESCRIPTIONS["get_profile"],
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": PARAMETER_DESCRIPTIONS["query"],
},
},
"required": [],
"additionalProperties": False,
},
},
"document_list": {
"name": "document_list",
"description": TOOL_DESCRIPTIONS["document_list"],
"parameters": {
"type": "object",
"properties": {
"limit": {
"type": "integer",
"description": PARAMETER_DESCRIPTIONS["limit"],
"default": DEFAULT_LIMIT,
"minimum": 1,
"maximum": 1100,
},
"page": {
"type": "integer",
"description": PARAMETER_DESCRIPTIONS["page"],
"default": 1,
"minimum": 1,
},
},
"required": [],
"additionalProperties": False,
},
},
"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"],
"additionalProperties": False,
},
},
"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"],
"additionalProperties": False,
},
},
"memory_forget": {
"name": "memory_forget",
"description": TOOL_DESCRIPTIONS["memory_forget"],
"parameters": {
"type": "object",
"properties": {
"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": [],
"additionalProperties": False,
},
},
}
def _resolve_container_tags(config: SupermemoryToolsConfig) -> List[str]:
project_id = config.get("project_id")
configured_tags = config.get("container_tags")
if project_id is not None and configured_tags is not None:
raise SupermemoryConfigurationError(
"Supermemory tools config accepts either project_id or container_tags, not both."
)
if project_id:
return [f"sm_project_{project_id}"]
if configured_tags is not None:
if not configured_tags or any(not tag for tag in configured_tags):
raise SupermemoryConfigurationError(
"container_tags must contain at least one non-empty tag."
)
return list(configured_tags)
return ["sm_project_default"]
def _tool_definition(name: str) -> ChatCompletionToolParam:
return {"type": "function", "function": MEMORY_TOOL_SCHEMAS[name]}
def _model_to_dict(value: Any) -> Dict[str, object]:
"""Normalize generated SDK models and already-plain response values."""
if isinstance(value, dict):
return dict(value)
model_dump = getattr(value, "model_dump", None)
if callable(model_dump):
dumped = model_dump()
if isinstance(dumped, dict):
return dumped
raise TypeError(f"Unsupported SDK response type: {type(value).__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."""
@ -109,54 +383,62 @@ class SupermemoryTools:
config: Optional configuration
"""
config = config or {}
# 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)
# 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"]
base_url = config.get("base_url")
if base_url:
self.client = supermemory.AsyncSupermemory(
api_key=api_key,
base_url=base_url,
)
else:
self.container_tags = ["sm_project_default"]
self.client = supermemory.AsyncSupermemory(api_key=api_key)
self.container_tags = _resolve_container_tags(config)
def _primary_container_tag(self) -> str:
return 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)
handlers: Dict[str, Any] = {
"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,
}
if function_name == "search_memories":
result = await self.search_memories(**args)
elif function_name == "add_memory":
result = await self.add_memory(**args)
else:
result = {
handler = handlers.get(function_name)
if handler is None:
result: Dict[str, object] = {
"success": False,
"error": f"Unknown function: {function_name}",
}
else:
try:
args = json.loads(tool_call.function.arguments)
except (json.JSONDecodeError, TypeError):
return json.dumps({"success": False, "error": "Invalid tool arguments"})
if not isinstance(args, dict):
return json.dumps({"success": False, "error": "Invalid tool arguments"})
parameters = MEMORY_TOOL_SCHEMAS[function_name]["parameters"] or {}
properties = parameters.get("properties", {})
required = parameters.get("required", [])
if not isinstance(properties, dict) or not isinstance(required, list):
return json.dumps({"success": False, "error": "Invalid tool arguments"})
required_names = {name for name in required if isinstance(name, str)}
if set(args) - set(properties) or required_names - set(args):
return json.dumps({"success": False, "error": "Invalid tool arguments"})
result = await handler(**args)
return json.dumps(result)
@ -164,18 +446,12 @@ class SupermemoryTools:
self,
information_to_get: str,
include_full_docs: Optional[bool] = None,
limit: int = 10,
limit: int = DEFAULT_LIMIT,
) -> MemorySearchResult:
"""Search memories.
Args:
information_to_get: Terms to search for
include_full_docs: Deprecated compatibility argument. V4 search
returns relevant memories and chunks, not full source documents.
limit: Maximum number of results
Returns:
MemorySearchResult
``include_full_docs`` remains a deprecated Python-only argument for
source compatibility. V4 search cannot return full source documents.
"""
if include_full_docs is not None:
warnings.warn(
@ -188,9 +464,9 @@ class SupermemoryTools:
try:
response: SearchMemoriesResponse = await self.client.search.memories(
q=information_to_get,
container_tag=self.container_tags[0],
container_tag=self._primary_container_tag(),
limit=limit,
threshold=0.6,
threshold=DEFAULT_CHUNK_THRESHOLD,
search_mode="hybrid",
)
@ -212,14 +488,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,
@ -241,32 +510,193 @@ class SupermemoryTools:
error=f"Memory add failed: {error}",
)
async def get_profile(
self,
query: Optional[str] = None,
) -> ProfileResult:
"""Get user profile with optional query-scoped search results."""
try:
if query:
response = await self.client.profile(
container_tag=self._primary_container_tag(),
q=query,
)
else:
response = await self.client.profile(
container_tag=self._primary_container_tag(),
)
return ProfileResult(
success=True,
profile=_model_to_dict(response.profile),
search_results=(
_model_to_dict(response.search_results)
if response.search_results is not None
else None
),
)
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,
limit: Optional[int] = None,
page: Optional[int] = None,
) -> DocumentListResult:
"""List stored documents."""
try:
kwargs: Dict[str, Any] = {
"container_tags": [self._primary_container_tag()],
"limit": DEFAULT_LIMIT if limit is None else limit,
}
if page is not None:
kwargs["page"] = page
response = await self.client.documents.list(**kwargs)
return DocumentListResult(
success=True,
documents=[_model_to_dict(document) for document in response.memories],
pagination=_model_to_dict(response.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:
# The delete endpoint has no container-tag argument. Resolve custom IDs
# first and refuse documents whose complete tag set is not configured.
document = await self.client.documents.get(document_id)
document_tags = set(document.container_tags or [])
configured_tags = set(self.container_tags)
if not document_tags or not document_tags.issubset(configured_tags):
return DocumentDeleteResult(
success=False,
error="Document is outside configured scope",
)
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,
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:
kwargs: Dict[str, Any] = {
"container_tag": self._primary_container_tag(),
}
if memory_id:
kwargs["id"] = memory_id
if memory_content:
kwargs["content"] = memory_content
if reason:
kwargs["reason"] = reason
await self.client.memories.forget(**kwargs)
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(
@ -274,16 +704,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(
@ -296,7 +717,6 @@ async def execute_memory_tool_calls(
content=result,
)
# Execute all tool calls concurrently
import asyncio
results = await asyncio.gather(
@ -306,22 +726,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: Optional[bool] = None,
limit: int = 10,
limit: int = DEFAULT_LIMIT,
) -> MemorySearchResult:
"""Execute search memories."""
return await self.tools.search_memories(
@ -336,41 +752,145 @@ 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,
query: Optional[str] = None,
) -> ProfileResult:
"""Execute get profile."""
return await self.tools.get_profile(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,
limit: Optional[int] = None,
page: Optional[int] = None,
) -> DocumentListResult:
"""Execute document list."""
return await self.tools.document_list(
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,
memory_id: Optional[str] = None,
memory_content: Optional[str] = None,
reason: Optional[str] = None,
) -> MemoryForgetResult:
"""Execute memory forget."""
return await self.tools.memory_forget(
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)

87
packages/openai-sdk-python/tests/test_tools.py
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(
@ -239,6 +233,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."""

8
packages/openai-sdk-python/uv.lock generated
View file

@ -1355,7 +1355,7 @@ wheels = [
[[package]]
name = "supermemory"
version = "3.56.0"
version = "3.59.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "anyio" },
@ -1365,14 +1365,14 @@ dependencies = [
{ name = "sniffio" },
{ name = "typing-extensions" },
]
sdist = { url = "https://files.pythonhosted.org/packages/10/43/3a7619a697554555d37254bd1e63ff2bf1262bae8ab569eba57f823af3e4/supermemory-3.56.0.tar.gz", hash = "sha256:3cceb35465e79762c2213a56d2d17b38c554924ba19fd29e24ce09701f4b377d", size = 175386, upload-time = "2026-07-24T16:29:16.695Z" }
sdist = { url = "https://files.pythonhosted.org/packages/30/83/7db28873de639f4c4ac469266a2142d1f9def545a0cac2a023011df23601/supermemory-3.59.0.tar.gz", hash = "sha256:5efd5a5a087d552b0e00e739e4eeac21379d087bdfaa1c9f5995d11272b35bb3", size = 154326, upload-time = "2026-08-14T21:28:39.913Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/09/2a/0d8ac89c73f540caa38ced4df3b824abae11481243c60f6ef750103c973d/supermemory-3.56.0-py3-none-any.whl", hash = "sha256:334e5cd1a8743ed9b90aa25ad0e63266f6376b4d44d5bdfc5cc4f15fc19d72f9", size = 156297, upload-time = "2026-07-24T16:29:15.491Z" },
{ url = "https://files.pythonhosted.org/packages/af/73/0cf59b31317baf66b0bb9d1fa6506a54013e49fb35ffc9c12c3f3a189c5c/supermemory-3.59.0-py3-none-any.whl", hash = "sha256:3c66ae0fcb082241d8a600b4f8aaebe68159aee7cd24c2ce1aae533ea404bb2d", size = 142020, upload-time = "2026-08-14T21:28:38.842Z" },
]
[[package]]
name = "supermemory-openai-sdk"
version = "1.0.6"
version = "1.0.7"
source = { editable = "." }
dependencies = [
{ name = "openai" },