ShortTermMemory API Reference

API for conversation and message storage.

Overview

ShortTermMemory handles conversations and messages with sequential linking, semantic search, and session management.

async with MemoryClient(settings) as client:
    # Access via client
    short_term = client.short_term

    # Add a message
    message = await short_term.add_message(
        session_id="user-123",
        role="user",
        content="Hello, world!"
    )

Message Operations

add_message

Add a message to a conversation.

message = await short_term.add_message(
    session_id: str,
    role: str,
    content: str,
    metadata: dict[str, Any] | None = None,
    timestamp: datetime | None = None,
)

Parameters

Parameter Type Default Description

session_id

str

required

Conversation/session identifier

role

str

required

Message role (user, assistant, system)

content

str

required

Message content

metadata

dict

None

Additional metadata

timestamp

datetime

now

Message timestamp

Returns

Message object with generated id and embedding.

add_messages

Add multiple messages in batch.

messages = await short_term.add_messages(
    session_id: str,
    messages: list[dict],  # [{"role": "user", "content": "..."}]
)

get_message

Get a message by ID.

message = await short_term.get_message(message_id: str)

get_messages

Get messages from a conversation.

messages = await short_term.get_messages(
    session_id: str,
    limit: int = 50,
    before: datetime | None = None,
    after: datetime | None = None,
)

delete_message

Delete a message.

await short_term.delete_message(message_id: str)

Search Operations

search_messages

Semantic search across messages.

results = await short_term.search_messages(
    query: str,
    session_id: str | None = None,
    limit: int = 10,
    min_score: float = 0.0,
    metadata_filter: dict[str, Any] | None = None,
)

Parameters

Parameter Type Default Description

query

str

required

Search query

session_id

str

None

Filter to specific session

limit

int

10

Maximum results

min_score

float

0.0

Minimum similarity score

metadata_filter

dict

None

Filter by metadata fields

Returns

List of (Message, score) tuples, sorted by relevance.

Session Operations

list_sessions

List all conversation sessions.

sessions = await short_term.list_sessions(
    user_id: str | None = None,
    limit: int = 100,
)

Returns

List of session info dictionaries:

[
    {
        "session_id": "user-123",
        "message_count": 42,
        "first_message": datetime(...),
        "last_message": datetime(...),
    },
    ...
]

get_session_messages

Get all messages in a session.

messages = await short_term.get_session_messages(
    session_id: str,
    limit: int | None = None,
)

delete_session

Delete a session and all its messages.

deleted_count = await short_term.delete_session(session_id: str)

Conversation Summaries

generate_summary

Generate an LLM summary of a conversation.

summary = await short_term.generate_summary(
    session_id: str,
    max_messages: int = 50,
)

Returns

ConversationSummary object:

@dataclass
class ConversationSummary:
    session_id: str
    summary: str
    message_count: int
    generated_at: datetime

get_summary

Get cached summary for a session.

summary = await short_term.get_summary(session_id: str)

Message Linking

Messages are automatically linked in sequence:

(Conversation) -[:FIRST_MESSAGE]-> (Message1)
(Message1) -[:NEXT_MESSAGE]-> (Message2)
(Message2) -[:NEXT_MESSAGE]-> (Message3)

migrate_message_links

Migrate existing data to sequential links (one-time operation).

stats = await short_term.migrate_message_links()
# Returns: {"session-1": 10, "session-2": 5, ...}

Message Model

@dataclass
class Message:
    id: str
    session_id: str
    role: str
    content: str
    timestamp: datetime
    metadata: dict[str, Any]
    embedding: list[float] | None

Example

async with MemoryClient(settings) as client:
    # Store conversation
    await client.short_term.add_message("user-123", "user", "What's the weather?")
    await client.short_term.add_message("user-123", "assistant", "It's sunny today!")

    # Search across all conversations
    results = await client.short_term.search_messages("weather forecast")
    for message, score in results:
        print(f"[{score:.2f}] {message.role}: {message.content}")

    # Get recent messages for a session
    messages = await client.short_term.get_messages("user-123", limit=10)

See Also