Quick Reference Code Samples
This page collects small, focused examples for common Oracle Agent Memory setup and API lifecycle operations.
LLM / Embedding Setup
The following examples use LiteLLM for both the LLM and embedding model.
Configuring an LLM
import oracledb
from oracleagentmemory.core.llms.llm import Llm
llm = Llm(
model="YOUR_LLM_MODEL",
api_base="YOUR_LLM_API_BASE",
api_key="YOUR_LLM_API_KEY",
)
#Call ``llm.generate("What is 2+2?")`` when the application needs a response.| API Reference: Llm | LlmResponse |
Configuring an Embedding Model
from oracleagentmemory.core.embedders.embedder import Embedder
embedder = Embedder(
model="YOUR_EMBEDDING_MODEL",
api_base="YOUR_EMBEDDING_API_BASE",
api_key="YOUR_EMBEDDING_API_KEY",
)
embedding_matrix = embedder.embed(["The quick brown fox jumps over the lazy dog"])
print(embedding_matrix.shape)
#(1, embedding_dimension)API Reference: Embedder
Configuring custom TLS certificates
When an OpenAI-compatible endpoint uses a private certificate authority, pass
the path to its PEM-encoded CA certificate or bundle as ca_file. If the
endpoint requires mutual TLS, also pass the client certificate chain and its
private key through cert_file and key_file. The client certificate and
private key must be provided together.
The paths are local filesystem paths. When these parameters are omitted, the
default TLS trust configuration is used. TLS configuration applies to the
provider-backed Llm and Embedder clients; it does not configure an
OracleDBEmbedder or the Oracle AI Database connection.
Provider requests honor proxy and TLS-related environment variables by default.
Use proxy to select an explicit proxy; it takes precedence over proxy
environment variables. Set trust_env=False to ignore those environment
settings. An explicitly configured proxy is still used when
trust_env=False.
from pathlib import Path
ca_file = Path("/path/to/provider-ca.pem")
if ca_file.is_file():
secure_llm = Llm(
model="openai/YOUR_LLM_MODEL",
api_base="https://YOUR_LLM_API_BASE/v1",
api_key="YOUR_LLM_API_KEY",
ca_file=str(ca_file),
proxy="http://proxy.example.com:8080",
trust_env=True,
)
client_cert_file = Path("/path/to/client-cert.pem")
client_key_file = Path("/path/to/client-key.pem")
if client_cert_file.is_file() and client_key_file.is_file():
mtls_embedder = Embedder(
model="hosted_vllm/YOUR_EMBEDDING_MODEL",
api_base="https://YOUR_EMBEDDING_API_BASE/v1",
api_key="YOUR_EMBEDDING_API_KEY",
ca_file=str(ca_file),
cert_file=str(client_cert_file),
key_file=str(client_key_file),
)
#Replace the example paths with files available in your deployment.
#``ca_file`` configures the CA used to verify the server. For mutual TLS,
#provide both ``cert_file`` and ``key_file`` as well.| API Reference: Llm | Embedder |
API Setup
Configuring an Agent Memory Component
This uses an Oracle DB connection or pool together with the embedding model and an optional LLM for automatic memory extraction.
import oracledb
from oracleagentmemory.core import SchemaPolicy
from oracleagentmemory.core.oracleagentmemory import OracleAgentMemory
db_pool = oracledb.SessionPool(
user="YOUR DB USER",
password="YOUR DB PASSWORD",
dsn="localhost:1521/...",
)
memory_store_id = "T_REF_SHEET"
memory_client = OracleAgentMemory(
connection=db_pool,
embedder=embedder,
llm=llm, # optional: enables automatic memory extraction during add_messages()
schema_policy=SchemaPolicy.CREATE_IF_NECESSARY,
memory_store_id=memory_store_id,
)| API Reference: OracleAgentMemory | Embedder |
Configuring an Oracle DB Memory Component
This variant uses an Oracle DB connection or pool and shows how to set a schema policy and a table name prefix.
import oracledb
from oracleagentmemory.core import SchemaPolicy
from oracleagentmemory.core.oracleagentmemory import OracleAgentMemory
db_pool = oracledb.SessionPool(
user="YOUR DB USER",
password="YOUR DB PASSWORD",
dsn="localhost:1521/...",
)
memory_store_id = "T_REF_SHEET"
memory = OracleAgentMemory(
connection=db_pool,
embedder=embedder,
llm=llm,
schema_policy=SchemaPolicy.CREATE_IF_NECESSARY,
memory_store_id=memory_store_id,
)| API Reference: SchemaPolicy | MemoryExtractionConfig |
Configuring an Oracle Hybrid DB Memory Component
This variant enables Oracle-managed hybrid search over stored search text and shows how to select the managed search-index sync mode.
SearchStrategy.HYBRID creates or validates Oracle’s managed hybrid vector
index and requires the main embedder to be an OracleDBEmbedder so the
managed index uses the embedder’s in-database model.
SearchStrategy.KEYWORD is text-only: it ranks by stored search text and
does not require an embedder. Keyword schemas can be created without local
vector storage, so do not reopen keyword schemas with SearchStrategy.VECTOR
unless you recreate the schema or backfill embeddings first. They can still be
upgraded to hybrid search with an OracleDBEmbedder because Oracle’s managed
hybrid index builds from stored search text.
Warning: When a hybrid index is first created over existing data, Oracle scans
stored search text and builds the managed index state during schema
setup. SchemaPolicy.CREATE_IF_NECESSARY can take time and should be
planned like a database migration for large schemas.
from oracleagentmemory.core import SchemaPolicy, SearchIndexSyncMode, SearchStrategy
from oracleagentmemory.core.embedders import OracleDBEmbedder
from oracleagentmemory.core.oracleagentmemory import OracleAgentMemory
db_embedder = OracleDBEmbedder(
connection=db_pool,
model="YOUR_DB_EMBEDDING_MODEL",
embedding_dimension=384,
)
hybrid_db_memory = OracleAgentMemory(
connection=db_pool,
embedder=db_embedder,
llm=llm,
schema_policy=SchemaPolicy.CREATE_IF_NECESSARY,
search_strategy=SearchStrategy.HYBRID,
search_index_sync=SearchIndexSyncMode.ON_COMMIT,
memory_store_id=memory_store_id,
)| API Reference: OracleAgentMemory | OracleDBEmbedder | SchemaPolicy | SearchStrategy | SearchIndexSyncMode |
API Lifecycle
Create a Thread
Create a thread with an optional thread ID, user ID, and agent ID.
thread = memory.create_thread(
thread_id="thread_create_123", # optional
user_id="user_123", # optional
agent_id="agent_456", # optional
)
print(thread.thread_id)
#thread_create_123API Reference: OracleThread
Re-open an Existing Thread
thread = memory.create_thread(
thread_id="thread_reopen_123",
user_id="user_123",
agent_id="agent_456",
)
same_thread = memory.get_thread("thread_reopen_123")
print(same_thread.thread_id)
#thread_reopen_123Update an Existing Thread
Use update_thread() to persist thread metadata or durable runtime-config
changes. Overrides passed to get_thread() affect only the reopened handle
until they are explicitly persisted.
thread = memory.create_thread(
thread_id="thread_update_123",
user_id="user_123",
agent_id="agent_456",
)
loaded_thread = memory.get_thread(
"thread_update_123",
max_message_token_length=8_000,
)
print(loaded_thread.max_message_token_length)
#8000
updated_thread = memory.update_thread(
"thread_update_123",
metadata={"source": "support", "flags": {"vip": True}},
max_message_token_length=8_000,
)
persisted_thread = memory.get_thread("thread_update_123")
print(updated_thread.metadata["flags"]["vip"])
#True
print(persisted_thread.max_message_token_length)
#8000
#Overrides passed to get_thread() are temporary. Call update_thread()
#to persist thread metadata or durable runtime-config changes.Delete a Thread
Use this operation when you need thread-scoped cascading cleanup. It removes the thread together with associated messages, durable memories, and backing retrieval data managed by the SDK.
thread = memory.create_thread(thread_id="thread_delete_123")
deleted = memory.delete_thread("thread_delete_123")
print(deleted)
#1
#Use thread deletion when you need thread-scoped cascading cleanup.
#It removes the thread together with its messages, memories,
#and backing retrieval data managed by the SDK.Add a User Profile
user_profile_id = memory.add_user(
"user_123",
"The user prefers concise answers and works mostly with Python.",
)
print(user_profile_id)
#user_123Add an Agent Profile
agent_profile_id = memory.add_agent(
"agent_456",
"A coding assistant specialized in debugging and code review.",
)
print(agent_profile_id)
#agent_456Add a Global Memory from the Memory API
When thread_id is omitted, the memory is not tied to a specific
thread. Use memory_type to store a general "memory" (the default),
"fact", "guideline", or "preference". The returned value is the
memory identifier.
memory_id = memory.add_memory(
"The user prefers short, bullet-point answers.",
memory_type="preference",
user_id="user_123",
agent_id="agent_456",
)
print(memory_id)
#mem:1Add a Scoped Memory from the Memory API
The returned value is the memory identifier. A record’s type is selected when
it is created and cannot be changed with update_memory().
thread = memory.create_thread(
thread_id="thread_scoped_123",
user_id="user_123",
agent_id="agent_456",
)
memory_id = memory.add_memory(
"The user is planning a trip to Kyoto next month.",
memory_type="fact",
user_id="user_123",
agent_id="agent_456",
thread_id=thread.thread_id,
)
print(memory_id)
#mem:2Update a Memory from the Memory API
Use update_memory() to replace stored content or metadata for an existing
memory-like record by identifier.
thread = memory.create_thread(
thread_id="thread_update_memory_api_123",
user_id="user_123",
agent_id="agent_456",
)
memory_id = memory.add_memory(
"The user likes short status updates.",
user_id=thread.user_id,
agent_id=thread.agent_id,
thread_id=thread.thread_id,
metadata={"source": "chat"},
)
updated_memory_id = memory.update_memory(
memory_id,
content="The user prefers short status updates.",
metadata={"source": "support"},
)
print(updated_memory_id)
#mem:3Add a Memory with a Custom ID
The returned value is the caller-provided memory identifier.
memory_id = memory.add_memory(
"The user prefers aisle seats on flights.",
user_id="user_123",
agent_id="agent_456",
memory_id="travel_pref_001",
)
print(memory_id)
#travel_pref_001Thread Basics
Add Messages to a Thread
Messages can be passed either as dictionaries or as Message objects.
Optional message IDs, timestamps, and metadata can be stored with them.
from oracleagentmemory.apis import Message
thread = memory.create_thread(
thread_id="thread_messages_123",
user_id="user_123",
agent_id="agent_456",
)
message_ids = thread.add_messages(
[
Message(
id="msg_user_001",
role="user",
content="I prefer window seats on flights.",
timestamp="2026-03-27T09:00:00Z",
metadata={"source": "assistant"},
),
{
"id": "msg_assistant_001",
"role": "assistant",
"content": "Noted. I will keep that in mind.",
"timestamp": "2026-03-27T09:00:05Z",
"metadata": {"source": "assistant"},
#message metadata must be identical for one `add_messages` call
},
]
)
print(message_ids)
#['msg_user_001', 'msg_assistant_001']API Reference: Message
Read Back Thread Messages
You can read all stored messages, or a slice using start and end.
thread = memory.create_thread(thread_id="thread_read_messages_123")
thread.add_messages(
[
{"role": "user", "content": "Message 1"},
{"role": "assistant", "content": "Message 2"},
{"role": "user", "content": "Message 3"},
]
)
default_messages = thread.get_messages()
all_messages = thread.get_messages(end=None)
middle_messages = thread.get_messages(start=1, end=3)
print([message.content for message in default_messages])
#On short threads, the bounded default still returns all messages.
#['Message 1', 'Message 2', 'Message 3']
print([message.content for message in all_messages])
#['Message 1', 'Message 2', 'Message 3']
print([message.content for message in middle_messages])
#['Message 2', 'Message 3']Delete a Message from the Current Thread by ID
Deleting a message removes only the raw message row from the current thread.
Derived memories or other downstream artifacts created from that message can
remain searchable and may still influence context-card output. If you need to
delete the thread together with its associated messages and memories, use
delete_thread() instead. Passing an identifier from a different thread
still returns 0.
thread = memory.create_thread(thread_id="thread_delete_message_123")
message_ids = thread.add_messages(
[
{"role": "user", "content": "Message to delete"},
]
)
deleted = thread.delete_message(message_ids[0])
print(deleted)
#1
#This removes only the raw message row from the current thread.
#Derived memories or other downstream artifacts created from that message
#are not deleted automatically and may remain searchable or appear in
#context-card output. Use memory.delete_thread(thread.thread_id) to delete
#the thread together with its associated messages and memories.
#Message deletes return 0 for IDs owned by another thread.Update a Message from the Current Thread by ID
Thread-scoped message updates affect only raw messages owned by the current
thread. Stored role and timestamp values are preserved, and, when automatic
extraction is enabled, editing message content immediately reruns extraction
for the edited message using the same history-window rules as
add_messages(). Only earlier thread history can be used as supporting
context. Later messages are ignored during that immediate pass, and the same
refresh keeps existing derived memories in place while adding any new
memories from the edited content.
thread = memory.create_thread(thread_id="thread_update_message_123")
thread.add_messages(
[
{
"id": "msg_update_001",
"role": "user",
"content": "Original message text.",
"timestamp": "2026-03-27T10:00:00Z",
"metadata": {"source": "chat"},
}
]
)
updated_message_id = thread.update_message(
"msg_update_001",
content="Edited message text.",
metadata={"source": "support"},
)
print(updated_message_id)
#msg_update_001
#Message updates preserve stored role and timestamp values.
#When automatic extraction is enabled, content edits immediately rerun
#extraction for the edited message using the same history-window
#rules as add_messages().
#Later messages are ignored during that immediate pass.
#Existing derived memories stay in place while new edited-content
#memories are added during that refresh.Add a Memory from a Thread Handle
The returned value is the memory identifier. This example stores a reusable
guideline; use memory_type="fact" or memory_type="preference" for
those categories, or omit it for a general memory.
thread = memory.create_thread(
thread_id="thread_add_memory_123",
user_id="user_123",
agent_id="agent_456",
)
memory_id = thread.add_memory(
"Use pytest for this repository's test suite.",
memory_type="guideline",
)
print(memory_id)
#mem:4Update a Memory from the Current Thread by ID
Thread-scoped updates affect only memory-like records owned by the current
thread. Passing an identifier from another thread raises KeyError.
thread = memory.create_thread(
thread_id="thread_update_memory_123",
user_id="user_123",
agent_id="agent_456",
)
memory_id = thread.add_memory(
"The user likes jasmine tea.",
metadata={"source": "survey"},
)
updated_memory_id = thread.update_memory(
memory_id,
content="The user likes jasmine tea in the afternoon.",
metadata={"source": "support"},
)
print(updated_memory_id)
#mem:5
#Thread updates are scoped to the current thread and raise KeyError
#for missing IDs or IDs owned by another thread.Delete a Memory from the Current Thread by ID
Thread deletes are scoped to the current thread. Passing an identifier from a
different thread returns 0.
thread = memory.create_thread(thread_id="thread_delete_memory_123")
memory_id = thread.add_memory("Temporary memory to delete.")
deleted = thread.delete_memory(memory_id)
print(deleted)
#1
#Thread deletes are scoped to the current thread and return 0 for IDs owned by another thread.Build a Context Card
thread = memory.create_thread(thread_id="thread_context_card_123")
thread.add_messages(
[
{"role": "user", "content": "I am planning a trip to Kyoto next spring."},
]
)
thread.add_memory("The user is planning a trip to Kyoto.")
context_card = thread.get_context_card()
print(context_card.content)
#<context_card>
#The user is planning a trip to Kyoto.
#</context_card>Build a Thread Summary
thread = memory.create_thread(thread_id="thread_summary_123")
thread.add_messages(
[
{"role": "user", "content": "Hello"},
{"role": "assistant", "content": "Hi, how can I help?"},
{"role": "user", "content": "Please summarize this thread."},
]
)
summary = thread.get_summary()
print(summary.content)
#user (-): Hello
#- assistant (-): Hi, how can I help?
#- user (-): Please summarize this thread.Build a Summary Excluding the Last N Messages
thread = memory.create_thread(thread_id="thread_summary_except_last_123")
thread.add_messages(
[
{"role": "user", "content": "First message"},
{"role": "assistant", "content": "Second message"},
{"role": "user", "content": "Third message"},
]
)
summary = thread.get_summary(except_last=1)
print(summary.content)
#user (-): First message
#- assistant (-): Second messageBuild a Summary with a Token Budget
thread = memory.create_thread(thread_id="thread_summary_budget_123")
thread.add_messages(
[
{"role": "user", "content": "Message 1"},
{"role": "assistant", "content": "Message 2"},
{"role": "user", "content": "Message 3"},
{"role": "assistant", "content": "Message 4"},
]
)
summary = thread.get_summary(token_budget=20)
print(summary.content)
#(truncated)
#user (-): Message 1
#...Search
Search from a Thread with No Explicit Scoping
Thread-level search uses the thread defaults when you do not pass an explicit scope.
thread = memory.create_thread(
thread_id="thread_search_default_123",
user_id="user_123",
agent_id="agent_456",
)
thread.add_memory("The user likes pizza.")
thread.add_memory("The user likes cats.")
results = thread.search("pizza", max_results=5)
print([result.content for result in results])
#['The user likes pizza.']API Reference: OracleSearchResult
Search from the Memory API with Scoping
At API level, you can scope retrieval with user_id, agent_id, and
thread_id through SearchScope.
For top-level client searches, provide an explicit user scope.
Use user_id=None only when you intentionally want unscoped records.
For a summary of how omitted values, explicit None, and exact-match flags
resolve at each API layer, see Scope Resolution.
from oracleagentmemory.apis.searchscope import SearchScope
thread = memory.create_thread(
thread_id="thread_memory_search_123",
user_id="user_123",
agent_id="agent_456",
)
thread.add_memory("The user likes hiking in the Alps.")
results = memory.search(
"hiking",
scope=SearchScope(
user_id="user_123",
agent_id="agent_456",
thread_id="thread_memory_search_123",
exact_thread_match=True,
),
max_results=5,
)
print([result.content for result in results])
#['The user likes hiking in the Alps.']API Reference: SearchScope
Search with Metadata Filtering
Use metadata_filter when search should only consider records whose stored
metadata contains a requested partial mapping. Multiple filter keys are
combined with AND semantics, nested dictionaries match nested metadata fields,
and bare list values must match exactly. To test array membership, use a
field-level operator dictionary such as
{"tags": {"$array_contains": "outdoor"}}. "$array_contains" with a
list requires all listed values, "$array_contains_any" requires at least
one listed value, and "$not" negates another field-level expression at the
same field, including an operator dictionary or raw exact-match value. A
negated expression also matches when the positive expression would fail,
including missing fields. Negated array membership also matches non-array
fields.
from oracleagentmemory.apis.searchscope import SearchScope
thread = memory.create_thread(
thread_id="thread_metadata_filter_123",
user_id="user_123",
agent_id="agent_456",
)
thread.add_memory(
"The user likes alpine hiking.",
metadata={"source": "survey", "category": {"kind": "travel"}, "tags": ["outdoor"]},
)
thread.add_memory(
"The user likes indoor climbing.",
metadata={"source": "chat", "category": {"kind": "sports"}, "tags": ["indoor"]},
)
results = memory.search(
"hiking",
scope=SearchScope(user_id="user_123"),
max_results=5,
record_types=["memory"],
metadata_filter={"source": "survey"},
)
print([result.content for result in results])
#['The user likes alpine hiking.']
outdoor_results = memory.search(
"hiking",
scope=SearchScope(user_id="user_123"),
max_results=5,
record_types=["memory"],
metadata_filter={
"source": "survey",
"tags": {"$array_contains": "outdoor"},
},
)
print([result.content for result in outdoor_results])
#['The user likes alpine hiking.']Search Only Memories or Only Messages
Use record_types to restrict search results to specific stored record
types.
thread = memory.create_thread(thread_id="thread_entity_type_search_123")
thread.add_messages(
[
{"role": "user", "content": "I mentioned pizza in a message."},
]
)
thread.add_memory("The user likes pizza.")
memory_results = thread.search("pizza", max_results=5, record_types=["memory"])
message_results = thread.search("pizza", max_results=5, record_types=["message"])
print([result.content for result in memory_results])
#['The user likes pizza.']
print([result.content for result in message_results])
#['I mentioned pizza in a message.']Full Code
Copy the complete code that follows.
#Copyright © 2026 Oracle and/or its affiliates.
#This software is under the Apache License 2.0
#(LICENSE-APACHE or http://www.apache.org/licenses/LICENSE-2.0) or Universal Permissive License
#(UPL) 1.0 (LICENSE-UPL or https://oss.oracle.com/licenses/upl), at your option.
#Oracle Agent Memory Code Example - Reference Sheet
#--------------------------------------------------
##Configure a LiteLLM LLM
import oracledb
from oracleagentmemory.core.llms.llm import Llm
llm = Llm(
model="YOUR_LLM_MODEL",
api_base="YOUR_LLM_API_BASE",
api_key="YOUR_LLM_API_KEY",
)
#Call ``llm.generate("What is 2+2?")`` when the application needs a response.
##Configure a LiteLLM embedding model
from oracleagentmemory.core.embedders.embedder import Embedder
embedder = Embedder(
model="YOUR_EMBEDDING_MODEL",
api_base="YOUR_EMBEDDING_API_BASE",
api_key="YOUR_EMBEDDING_API_KEY",
)
embedding_matrix = embedder.embed(["The quick brown fox jumps over the lazy dog"])
print(embedding_matrix.shape)
#(1, embedding_dimension)
##Configure custom TLS for OpenAI-compatible providers
from pathlib import Path
ca_file = Path("/path/to/provider-ca.pem")
if ca_file.is_file():
secure_llm = Llm(
model="openai/YOUR_LLM_MODEL",
api_base="https://YOUR_LLM_API_BASE/v1",
api_key="YOUR_LLM_API_KEY",
ca_file=str(ca_file),
proxy="http://proxy.example.com:8080",
trust_env=True,
)
client_cert_file = Path("/path/to/client-cert.pem")
client_key_file = Path("/path/to/client-key.pem")
if client_cert_file.is_file() and client_key_file.is_file():
mtls_embedder = Embedder(
model="hosted_vllm/YOUR_EMBEDDING_MODEL",
api_base="https://YOUR_EMBEDDING_API_BASE/v1",
api_key="YOUR_EMBEDDING_API_KEY",
ca_file=str(ca_file),
cert_file=str(client_cert_file),
key_file=str(client_key_file),
)
#Replace the example paths with files available in your deployment.
#``ca_file`` configures the CA used to verify the server. For mutual TLS,
#provide both ``cert_file`` and ``key_file`` as well.
##Configure an Oracle Memory component
from oracleagentmemory.core import SchemaPolicy
from oracleagentmemory.core.oracleagentmemory import OracleAgentMemory
db_pool = oracledb.SessionPool(
user="YOUR DB USER",
password="YOUR DB PASSWORD",
dsn="localhost:1521/...",
)
memory_store_id = "T_REF_SHEET"
memory_client = OracleAgentMemory(
connection=db_pool,
embedder=embedder,
llm=llm, # optional: enables automatic memory extraction during add_messages()
schema_policy=SchemaPolicy.CREATE_IF_NECESSARY,
memory_store_id=memory_store_id,
)
##Configure an Oracle DB component
from oracleagentmemory.core.oracleagentmemory import OracleAgentMemory
db_pool = oracledb.SessionPool(
user="YOUR DB USER",
password="YOUR DB PASSWORD",
dsn="localhost:1521/...",
)
memory_store_id = "T_REF_SHEET"
memory = OracleAgentMemory(
connection=db_pool,
embedder=embedder,
llm=llm,
schema_policy=SchemaPolicy.CREATE_IF_NECESSARY,
memory_store_id=memory_store_id,
)
##Configure an Oracle Hybrid DB component
from oracleagentmemory.core import SchemaPolicy, SearchIndexSyncMode, SearchStrategy
from oracleagentmemory.core.embedders import OracleDBEmbedder
from oracleagentmemory.core.oracleagentmemory import OracleAgentMemory
db_embedder = OracleDBEmbedder(
connection=db_pool,
model="YOUR_DB_EMBEDDING_MODEL",
embedding_dimension=384,
)
hybrid_db_memory = OracleAgentMemory(
connection=db_pool,
embedder=db_embedder,
llm=llm,
schema_policy=SchemaPolicy.CREATE_IF_NECESSARY,
search_strategy=SearchStrategy.HYBRID,
search_index_sync=SearchIndexSyncMode.ON_COMMIT,
memory_store_id=memory_store_id,
)
##Create a thread
thread = memory.create_thread(
thread_id="thread_create_123", # optional
user_id="user_123", # optional
agent_id="agent_456", # optional
)
print(thread.thread_id)
#thread_create_123
##Re open an existing thread
thread = memory.create_thread(
thread_id="thread_reopen_123",
user_id="user_123",
agent_id="agent_456",
)
same_thread = memory.get_thread("thread_reopen_123")
print(same_thread.thread_id)
#thread_reopen_123
##Update an existing thread
thread = memory.create_thread(
thread_id="thread_update_123",
user_id="user_123",
agent_id="agent_456",
)
loaded_thread = memory.get_thread(
"thread_update_123",
max_message_token_length=8_000,
)
print(loaded_thread.max_message_token_length)
#8000
updated_thread = memory.update_thread(
"thread_update_123",
metadata={"source": "support", "flags": {"vip": True}},
max_message_token_length=8_000,
)
persisted_thread = memory.get_thread("thread_update_123")
print(updated_thread.metadata["flags"]["vip"])
#True
print(persisted_thread.max_message_token_length)
#8000
#Overrides passed to get_thread() are temporary. Call update_thread()
#to persist thread metadata or durable runtime-config changes.
##Delete a thread
thread = memory.create_thread(thread_id="thread_delete_123")
deleted = memory.delete_thread("thread_delete_123")
print(deleted)
#1
#Use thread deletion when you need thread-scoped cascading cleanup.
#It removes the thread together with its messages, memories,
#and backing retrieval data managed by the SDK.
##Add a user profile
user_profile_id = memory.add_user(
"user_123",
"The user prefers concise answers and works mostly with Python.",
)
print(user_profile_id)
#user_123
##Add an agent profile
agent_profile_id = memory.add_agent(
"agent_456",
"A coding assistant specialized in debugging and code review.",
)
print(agent_profile_id)
#agent_456
##Add a global memory from the memory API
memory_id = memory.add_memory(
"The user prefers short, bullet-point answers.",
memory_type="preference",
user_id="user_123",
agent_id="agent_456",
)
print(memory_id)
#mem:1
##Add a scoped memory from the memory API
thread = memory.create_thread(
thread_id="thread_scoped_123",
user_id="user_123",
agent_id="agent_456",
)
memory_id = memory.add_memory(
"The user is planning a trip to Kyoto next month.",
memory_type="fact",
user_id="user_123",
agent_id="agent_456",
thread_id=thread.thread_id,
)
print(memory_id)
#mem:2
##Update a memory from the memory API
thread = memory.create_thread(
thread_id="thread_update_memory_api_123",
user_id="user_123",
agent_id="agent_456",
)
memory_id = memory.add_memory(
"The user likes short status updates.",
user_id=thread.user_id,
agent_id=thread.agent_id,
thread_id=thread.thread_id,
metadata={"source": "chat"},
)
updated_memory_id = memory.update_memory(
memory_id,
content="The user prefers short status updates.",
metadata={"source": "support"},
)
print(updated_memory_id)
#mem:3
##Add a memory with a custom ID
memory_id = memory.add_memory(
"The user prefers aisle seats on flights.",
user_id="user_123",
agent_id="agent_456",
memory_id="travel_pref_001",
)
print(memory_id)
#travel_pref_001
##Add messages to a thread
from oracleagentmemory.apis import Message
thread = memory.create_thread(
thread_id="thread_messages_123",
user_id="user_123",
agent_id="agent_456",
)
message_ids = thread.add_messages(
[
Message(
id="msg_user_001",
role="user",
content="I prefer window seats on flights.",
timestamp="2026-03-27T09:00:00Z",
metadata={"source": "assistant"},
),
{
"id": "msg_assistant_001",
"role": "assistant",
"content": "Noted. I will keep that in mind.",
"timestamp": "2026-03-27T09:00:05Z",
"metadata": {"source": "assistant"},
#message metadata must be identical for one `add_messages` call
},
]
)
print(message_ids)
#['msg_user_001', 'msg_assistant_001']
##Read back thread messages
thread = memory.create_thread(thread_id="thread_read_messages_123")
thread.add_messages(
[
{"role": "user", "content": "Message 1"},
{"role": "assistant", "content": "Message 2"},
{"role": "user", "content": "Message 3"},
]
)
default_messages = thread.get_messages()
all_messages = thread.get_messages(end=None)
middle_messages = thread.get_messages(start=1, end=3)
print([message.content for message in default_messages])
#On short threads, the bounded default still returns all messages.
#['Message 1', 'Message 2', 'Message 3']
print([message.content for message in all_messages])
#['Message 1', 'Message 2', 'Message 3']
print([message.content for message in middle_messages])
#['Message 2', 'Message 3']
##Delete a message from the current thread by ID
thread = memory.create_thread(thread_id="thread_delete_message_123")
message_ids = thread.add_messages(
[
{"role": "user", "content": "Message to delete"},
]
)
deleted = thread.delete_message(message_ids[0])
print(deleted)
#1
#This removes only the raw message row from the current thread.
#Derived memories or other downstream artifacts created from that message
#are not deleted automatically and may remain searchable or appear in
#context-card output. Use memory.delete_thread(thread.thread_id) to delete
#the thread together with its associated messages and memories.
#Message deletes return 0 for IDs owned by another thread.
##Update a message from the current thread by ID
thread = memory.create_thread(thread_id="thread_update_message_123")
thread.add_messages(
[
{
"id": "msg_update_001",
"role": "user",
"content": "Original message text.",
"timestamp": "2026-03-27T10:00:00Z",
"metadata": {"source": "chat"},
}
]
)
updated_message_id = thread.update_message(
"msg_update_001",
content="Edited message text.",
metadata={"source": "support"},
)
print(updated_message_id)
#msg_update_001
#Message updates preserve stored role and timestamp values.
#When automatic extraction is enabled, content edits immediately rerun
#extraction for the edited message using the same history-window
#rules as add_messages().
#Later messages are ignored during that immediate pass.
#Existing derived memories stay in place while new edited-content
#memories are added during that refresh.
##Add a memory from a thread handle
thread = memory.create_thread(
thread_id="thread_add_memory_123",
user_id="user_123",
agent_id="agent_456",
)
memory_id = thread.add_memory(
"Use pytest for this repository's test suite.",
memory_type="guideline",
)
print(memory_id)
#mem:4
##Update a memory from the current thread by ID
thread = memory.create_thread(
thread_id="thread_update_memory_123",
user_id="user_123",
agent_id="agent_456",
)
memory_id = thread.add_memory(
"The user likes jasmine tea.",
metadata={"source": "survey"},
)
updated_memory_id = thread.update_memory(
memory_id,
content="The user likes jasmine tea in the afternoon.",
metadata={"source": "support"},
)
print(updated_memory_id)
#mem:5
#Thread updates are scoped to the current thread and raise KeyError
#for missing IDs or IDs owned by another thread.
##Delete a memory from the current thread by ID
thread = memory.create_thread(thread_id="thread_delete_memory_123")
memory_id = thread.add_memory("Temporary memory to delete.")
deleted = thread.delete_memory(memory_id)
print(deleted)
#1
#Thread deletes are scoped to the current thread and return 0 for IDs owned by another thread.
##Build a context card
thread = memory.create_thread(thread_id="thread_context_card_123")
thread.add_messages(
[
{"role": "user", "content": "I am planning a trip to Kyoto next spring."},
]
)
thread.add_memory("The user is planning a trip to Kyoto.")
context_card = thread.get_context_card()
print(context_card.content)
#<context_card>
#The user is planning a trip to Kyoto.
#</context_card>
##Build a thread summary
thread = memory.create_thread(thread_id="thread_summary_123")
thread.add_messages(
[
{"role": "user", "content": "Hello"},
{"role": "assistant", "content": "Hi, how can I help?"},
{"role": "user", "content": "Please summarize this thread."},
]
)
summary = thread.get_summary()
print(summary.content)
#user (-): Hello
#- assistant (-): Hi, how can I help?
#- user (-): Please summarize this thread.
##Build a summary excluding the last N messages
thread = memory.create_thread(thread_id="thread_summary_except_last_123")
thread.add_messages(
[
{"role": "user", "content": "First message"},
{"role": "assistant", "content": "Second message"},
{"role": "user", "content": "Third message"},
]
)
summary = thread.get_summary(except_last=1)
print(summary.content)
#user (-): First message
#- assistant (-): Second message
##Build a summary with a token budget
thread = memory.create_thread(thread_id="thread_summary_budget_123")
thread.add_messages(
[
{"role": "user", "content": "Message 1"},
{"role": "assistant", "content": "Message 2"},
{"role": "user", "content": "Message 3"},
{"role": "assistant", "content": "Message 4"},
]
)
summary = thread.get_summary(token_budget=20)
print(summary.content)
#(truncated)
#user (-): Message 1
#...
##Search from a thread with no explicit scoping
thread = memory.create_thread(
thread_id="thread_search_default_123",
user_id="user_123",
agent_id="agent_456",
)
thread.add_memory("The user likes pizza.")
thread.add_memory("The user likes cats.")
results = thread.search("pizza", max_results=5)
print([result.content for result in results])
#['The user likes pizza.']
##Search from the memory API with scoping
from oracleagentmemory.apis.searchscope import SearchScope
thread = memory.create_thread(
thread_id="thread_memory_search_123",
user_id="user_123",
agent_id="agent_456",
)
thread.add_memory("The user likes hiking in the Alps.")
results = memory.search(
"hiking",
scope=SearchScope(
user_id="user_123",
agent_id="agent_456",
thread_id="thread_memory_search_123",
exact_thread_match=True,
),
max_results=5,
)
print([result.content for result in results])
#['The user likes hiking in the Alps.']
##Search with metadata filtering
from oracleagentmemory.apis.searchscope import SearchScope
thread = memory.create_thread(
thread_id="thread_metadata_filter_123",
user_id="user_123",
agent_id="agent_456",
)
thread.add_memory(
"The user likes alpine hiking.",
metadata={"source": "survey", "category": {"kind": "travel"}, "tags": ["outdoor"]},
)
thread.add_memory(
"The user likes indoor climbing.",
metadata={"source": "chat", "category": {"kind": "sports"}, "tags": ["indoor"]},
)
results = memory.search(
"hiking",
scope=SearchScope(user_id="user_123"),
max_results=5,
record_types=["memory"],
metadata_filter={"source": "survey"},
)
print([result.content for result in results])
#['The user likes alpine hiking.']
outdoor_results = memory.search(
"hiking",
scope=SearchScope(user_id="user_123"),
max_results=5,
record_types=["memory"],
metadata_filter={
"source": "survey",
"tags": {"$array_contains": "outdoor"},
},
)
print([result.content for result in outdoor_results])
#['The user likes alpine hiking.']
##Search only memories or messages
thread = memory.create_thread(thread_id="thread_entity_type_search_123")
thread.add_messages(
[
{"role": "user", "content": "I mentioned pizza in a message."},
]
)
thread.add_memory("The user likes pizza.")
memory_results = thread.search("pizza", max_results=5, record_types=["memory"])
message_results = thread.search("pizza", max_results=5, record_types=["message"])
print([result.content for result in memory_results])
#['The user likes pizza.']
print([result.content for result in message_results])
#['I mentioned pizza in a message.']