Skip to content

Memory

aigise.memory.search.search_controller

Search controller for orchestrating memory searches.

STRATEGY_REGISTRY = {'embedding_search': EmbeddingSearchStrategy, 'keyword_search': KeywordSearchStrategy, 'title_browse': TitleBrowseStrategy} module-attribute

logger = logging.getLogger(__name__) module-attribute

DomainConfig dataclass

Configuration for a knowledge domain in the memory system.

A domain defines a coherent set of node types, relationships, and search strategies for a particular area of knowledge (e.g., code, Q&A pairs, documentation).

Multiple domains can be combined to create a rich knowledge graph.

default_strategy: str = 'embedding_search' class-attribute instance-attribute

Default search strategy if LLM doesn't select one.

description: str = '' class-attribute instance-attribute

Human-readable description of the domain.

embedding_dimension: int = 3072 class-attribute instance-attribute

Default embedding dimension for this domain (Gemini default).

name: str instance-attribute

Unique identifier for this domain.

node_types: Dict[str, NodeTypeConfig] = field(default_factory=dict) class-attribute instance-attribute

Node type configurations keyed by label.

relationships: Dict[str, RelationshipConfig] = field(default_factory=dict) class-attribute instance-attribute

Relationship configurations keyed by type name.

search_strategies: List[str] = field(default_factory=list) class-attribute instance-attribute

Ordered list of search strategy names to try.

__post_init__()

Validate and register the domain configuration.

get_node_labels() -> List[str]

Get all node labels in this domain.

get_node_type(label: str) -> Optional[NodeTypeConfig]

Get a node type configuration by label.

get_relationship(type_name: str) -> Optional[RelationshipConfig]

Get a relationship configuration by type name.

get_relationship_types() -> List[str]

Get all relationship types in this domain.

get_similarity_searchable_types() -> List[str]

Get node types that support similarity search.

merge_with(other: 'DomainConfig') -> 'DomainConfig'

Merge this domain with another, creating a combined configuration.

The other domain's configurations take precedence on conflicts.

Parameters:

Name Type Description Default
other 'DomainConfig'

Another domain configuration to merge with.

required

Returns:

Type Description
'DomainConfig'

New merged DomainConfig.

validate() -> List[str]

Validate the domain configuration.

Returns:

Type Description
List[str]

List of validation error messages (empty if valid).

MemorySearchController

Controller for orchestrating memory search operations.

This controller: 1. Selects appropriate search strategies (LLM-driven or heuristic) 2. Executes searches with multi-round refinement 3. Evaluates result sufficiency 4. Combines results from multiple strategies if needed

__init__(domain_config: Optional['DomainConfig'] = None, max_iterations: int = 3, sufficiency_threshold: int = 1, use_llm_selection: bool = True)

Initialize the search controller.

Parameters:

Name Type Description Default
domain_config Optional['DomainConfig']

Domain configuration defining available strategies.

None
max_iterations int

Maximum number of refinement iterations.

3
sufficiency_threshold int

Minimum results needed to consider sufficient.

1
use_llm_selection bool

Whether to use LLM for strategy selection.

True

add_strategy(name: str, strategy: SearchStrategy) -> None

Add a custom search strategy.

get_available_strategies() -> List[str]

Get names of available strategies.

search(query: str, node_types: Optional[List[str]] = None, client: Any = None, max_results: int = 10, min_score: float = 0.0, metadata: Optional[Dict[str, Any]] = None) -> SearchResult async

Execute a memory search.

Parameters:

Name Type Description Default
query str

The search query.

required
node_types Optional[List[str]]

Optional list of node types to search.

None
client Any

Neo4j client for executing queries.

None
max_results int

Maximum results to return.

10
min_score float

Minimum score threshold.

0.0
metadata Optional[Dict[str, Any]]

Additional search context.

None

Returns:

Type Description
SearchResult

SearchResult with found items and metadata.

SearchContext dataclass

Context for a search operation.

domain_config: Optional['DomainConfig'] = None class-attribute instance-attribute

Domain configuration to use.

max_results: int = 10 class-attribute instance-attribute

Maximum number of results to return.

metadata: Dict[str, Any] = field(default_factory=dict) class-attribute instance-attribute

Additional context metadata.

min_score: float = 0.0 class-attribute instance-attribute

Minimum score threshold.

node_types: Optional[List[str]] = None class-attribute instance-attribute

Restrict search to specific node types (None = all).

query: str instance-attribute

The search query.

SearchResult dataclass

Result of a memory search operation.

has_results: bool property

Check if any results were found.

items: List[SearchResultItem] = field(default_factory=list) class-attribute instance-attribute

Search result items.

iterations: int = 1 class-attribute instance-attribute

Number of search iterations performed.

metadata: Dict[str, Any] = field(default_factory=dict) class-attribute instance-attribute

Additional search metadata.

strategy_used: str = '' class-attribute instance-attribute

Name of the strategy that found results.

sufficient: bool = False class-attribute instance-attribute

Whether the results are considered sufficient.

total_found: int = 0 class-attribute instance-attribute

Total number of results found (before limiting).

get_best_result() -> Optional[SearchResultItem]

Get the highest-scoring result.

SearchResultItem dataclass

A single search result item.

highlight: Optional[str] = None class-attribute instance-attribute

Highlighted match context (for keyword search).

match_type: str = 'exact' class-attribute instance-attribute

Type of match: 'exact', 'similarity', 'keyword', 'browse'.

node_id: str instance-attribute

The Neo4j element ID of the matched node.

node_label: str instance-attribute

The Neo4j label of the matched node.

properties: Dict[str, Any] instance-attribute

Properties of the matched node.

score: float = 1.0 class-attribute instance-attribute

Relevance score (0.0 to 1.0).

get_display_text() -> str

Get a human-readable display text for this result.

SearchStrategy

Bases: ABC

Abstract base class for search strategies.

A search strategy defines how to search for relevant information in the memory graph. Different strategies may use vector similarity, keyword matching, graph traversal, etc.

can_handle_query(query: str, context: SearchContext) -> float async

Estimate how well this strategy can handle a query.

This method helps the search controller select the best strategy. Override to provide better estimates based on query characteristics.

Parameters:

Name Type Description Default
query str

The search query.

required
context SearchContext

Search context.

required

Returns:

Type Description
float

Confidence score from 0.0 (cannot handle) to 1.0 (ideal match).

get_description() -> str abstractmethod

Get a human-readable description of this strategy.

get_name() -> str abstractmethod

Get the unique name of this strategy.

get_supported_node_types() -> Optional[List[str]]

Get the list of node types this strategy supports.

Returns:

Type Description
Optional[List[str]]

List of supported node types, or None for all types.

search(context: SearchContext, client: Any) -> List[SearchResultItem] abstractmethod async

Execute the search strategy.

Parameters:

Name Type Description Default
context SearchContext

Search context with query and parameters.

required
client Any

Neo4j client for executing queries.

required

Returns:

Type Description
List[SearchResultItem]

List of search result items, ordered by relevance.

supports_node_type(node_type: str) -> bool

Check if this strategy supports a particular node type.

Override in subclasses to restrict to specific node types.

Parameters:

Name Type Description Default
node_type str

Node label to check.

required

Returns:

Type Description
bool

True if this strategy can search this node type.

get_memory_settings() -> MemorySettings

Get the global memory settings instance.

Returns:

Name Type Description
MemorySettings MemorySettings

The singleton settings instance.

aigise.memory.tools.memory_search_tools

Memory search tools for agents.

_search_controller: Optional[MemorySearchController] = None module-attribute

logger = logging.getLogger(__name__) module-attribute

MemorySearchController

Controller for orchestrating memory search operations.

This controller: 1. Selects appropriate search strategies (LLM-driven or heuristic) 2. Executes searches with multi-round refinement 3. Evaluates result sufficiency 4. Combines results from multiple strategies if needed

__init__(domain_config: Optional['DomainConfig'] = None, max_iterations: int = 3, sufficiency_threshold: int = 1, use_llm_selection: bool = True)

Initialize the search controller.

Parameters:

Name Type Description Default
domain_config Optional['DomainConfig']

Domain configuration defining available strategies.

None
max_iterations int

Maximum number of refinement iterations.

3
sufficiency_threshold int

Minimum results needed to consider sufficient.

1
use_llm_selection bool

Whether to use LLM for strategy selection.

True

add_strategy(name: str, strategy: SearchStrategy) -> None

Add a custom search strategy.

get_available_strategies() -> List[str]

Get names of available strategies.

search(query: str, node_types: Optional[List[str]] = None, client: Any = None, max_results: int = 10, min_score: float = 0.0, metadata: Optional[Dict[str, Any]] = None) -> SearchResult async

Execute a memory search.

Parameters:

Name Type Description Default
query str

The search query.

required
node_types Optional[List[str]]

Optional list of node types to search.

None
client Any

Neo4j client for executing queries.

None
max_results int

Maximum results to return.

10
min_score float

Minimum score threshold.

0.0
metadata Optional[Dict[str, Any]]

Additional search context.

None

Returns:

Type Description
SearchResult

SearchResult with found items and metadata.

_get_search_controller() -> MemorySearchController

Get or create the search controller singleton.

get_entity_by_id(node_label: str, node_key: Dict[str, Any], *, include_relationships: bool = False, tool_context: ToolContext) -> Dict[str, Any] async

Get a specific entity from memory by its identifier.

Parameters:

Name Type Description Default
node_label str

Label of the node (e.g., "Question", "Answer").

required
node_key Dict[str, Any]

Properties to identify the node (e.g., {"answer_id": "..."}).

required
include_relationships bool

Whether to include connected entities.

False

Returns:

Type Description
Dict[str, Any]

Dictionary with:

Dict[str, Any]
  • success: True if query completed
Dict[str, Any]
  • found: True if the entity exists
Dict[str, Any]
  • entity: The entity properties (if found)
Dict[str, Any]
  • relationships: Related entities (if include_relationships=True)

get_merged_domain(*domain_names: str) -> DomainConfig

Get a merged domain from multiple registered domains.

Parameters:

Name Type Description Default
*domain_names str

Names of domains to merge.

()

Returns:

Type Description
DomainConfig

Merged domain configuration.

Raises:

Type Description
ValueError

If any domain name is not found.

Get entities related to a specific node in the memory graph.

Use this to explore connections between cached knowledge, such as finding all topics related to a question or all content that references a specific topic.

Parameters:

Name Type Description Default
node_label str

Label of the node to find relations for (e.g., "Topic").

required
node_key Dict[str, Any]

Properties to identify the node (e.g., {"name": "authentication"}).

required
relationship_types Optional[List[str]]

Optional list of relationship types to follow. Options: "ABOUT", "HAS_ANSWER", "HAS_TOPIC", "RELATED_TO", "MENTIONS".

None
direction str

Direction to traverse relationships: - "outgoing": source -> target (what this node points to) - "incoming": source <- target (what points to this node) - "both": both directions (default)

'both'
max_results int

Maximum number of results. Default is 10.

10

Returns:

Type Description
Dict[str, Any]

Dictionary with:

Dict[str, Any]
  • success: True if query completed
Dict[str, Any]
  • source_found: True if the source node exists
Dict[str, Any]
  • relationships: Dict mapping relationship type to list of related nodes
Example

get_related_entities( node_label="Topic", node_key={"name": "authentication"}, relationship_types=["HAS_TOPIC"], direction="incoming", # Text --[HAS_TOPIC]--> Topic )

list_memory_contents(*, node_types: Optional[List[str]] = None, limit: int = 20, offset: int = 0, order_by: str = 'created_at', tool_context: ToolContext) -> Dict[str, Any] async

List contents of the memory graph by node type.

Use this to browse what's in memory without a specific search query. Useful for exploring available cached knowledge.

Parameters:

Name Type Description Default
node_types Optional[List[str]]

Node types to list. Defaults to ["Question", "Topic"].

None
limit int

Maximum items per type. Default is 20.

20
offset int

Number of items to skip (for pagination). Default is 0.

0
order_by str

Property to order by. Default is "created_at".

'created_at'

Returns:

Type Description
Dict[str, Any]

Dictionary with:

Dict[str, Any]
  • success: True if query completed
Dict[str, Any]
  • contents: Dict mapping node type to list of items
Dict[str, Any]
  • totals: Dict mapping node type to total count

requires_sandbox(*sandbox_types: str) -> Callable[[F], F]

Universal decorator for declaring sandbox dependencies.

This decorator works for both: - Tool functions: Only marks the function with metadata - Toolset factories: Marks function AND injects metadata into returned instance

The decorator is purely declarative - it does not create or fetch sandboxes. It only adds __sandbox_requirements__ metadata for static analysis via collect_sandbox_dependencies().

Parameters:

Name Type Description Default
*sandbox_types str

Variable number of sandbox type names that the tool or toolset depends on (e.g., "main", "gdb_mcp", "neo4j").

()

Returns:

Type Description
Callable[[F], F]

A decorator function that adds sandbox_requirements metadata.

safe_tool_execution(func: F) -> F

Decorator to wrap tool functions with error handling.

Catches all exceptions and returns a formatted error message with backtrace. Works for both sync and async functions.

Returns:

Type Description
F

dict with "error" key containing failure message and backtrace

search_memory(query: str, *, node_types: Optional[List[str]] = None, max_results: int = 5, min_score: float = 0.5, tool_context: ToolContext) -> Dict[str, Any] async

Search the memory graph for relevant cached knowledge.

Use this tool to find cached Q&A pairs, topics, or code entities that are semantically similar to your query.

Parameters:

Name Type Description Default
query str

The search query - describe what you're looking for.

required
node_types Optional[List[str]]

Optional list of node types to search. Defaults to ["Question", "Topic"]. Other options: "Answer", "Function", "Class", "File".

None
max_results int

Maximum number of results to return. Default is 5.

5
min_score float

Minimum similarity score (0-1). Default is 0.5.

0.5

Returns:

Type Description
Dict[str, Any]

Dictionary with:

Dict[str, Any]
  • success: True if search completed
Dict[str, Any]
  • found: True if any results were found
Dict[str, Any]
  • results: List of matching items with scores
Dict[str, Any]
  • best_match: The highest scoring result (if any)
Dict[str, Any]
  • strategy_used: Which search strategy found results

aigise.memory.tools.memory_update_tools

Memory update tools for agents.

_update_controller: Optional[MemoryUpdateController] = None module-attribute

logger = logging.getLogger(__name__) module-attribute

GraphOperations

Executes graph operations for adding, updating, and deleting nodes/relationships.

This class handles: - Creating new nodes with proper MERGE handling - Updating existing nodes - Creating relationships between nodes - Ensuring indexes exist

__init__(domain_config: Optional['DomainConfig'] = None)

Initialize graph operations.

Parameters:

Name Type Description Default
domain_config Optional['DomainConfig']

Domain configuration for node/relationship schemas.

None

add_entities_batch(entities: List[ExtractedEntity], client: Any, aigise_session_id: Optional[str] = None) -> List[OperationResult] async

Add multiple entities to the graph.

Parameters:

Name Type Description Default
entities List[ExtractedEntity]

Entities to add.

required
client Any

Neo4j client.

required
aigise_session_id Optional[str]

Optional session ID.

None

Returns:

Type Description
List[OperationResult]

List of operation results.

add_entity(entity: ExtractedEntity, client: Any, aigise_session_id: Optional[str] = None) -> OperationResult async

Add an entity to the graph.

Uses MERGE to avoid duplicates based on unique keys.

Parameters:

Name Type Description Default
entity ExtractedEntity

Entity to add.

required
client Any

Neo4j client.

required
aigise_session_id Optional[str]

Optional session ID for tracking.

None

Returns:

Type Description
OperationResult

OperationResult with operation details.

add_relationship(relationship: DiscoveredRelationship, client: Any) -> OperationResult async

Add a relationship to the graph.

Parameters:

Name Type Description Default
relationship DiscoveredRelationship

Relationship to add.

required
client Any

Neo4j client.

required

Returns:

Type Description
OperationResult

OperationResult with operation details.

add_relationships_batch(relationships: List[DiscoveredRelationship], client: Any) -> List[OperationResult] async

Add multiple relationships to the graph.

Parameters:

Name Type Description Default
relationships List[DiscoveredRelationship]

Relationships to add.

required
client Any

Neo4j client.

required

Returns:

Type Description
List[OperationResult]

List of operation results.

delete_entity(label: str, match_key: Dict[str, Any], client: Any) -> OperationResult async

Delete a node from the graph.

Uses DETACH DELETE to remove the node and all its relationships.

Parameters:

Name Type Description Default
label str

Node label (e.g., "Question", "Topic").

required
match_key Dict[str, Any]

Properties to identify the node to delete.

required
client Any

Neo4j client.

required

Returns:

Type Description
OperationResult

OperationResult with operation details.

delete_relationship(rel_type: str, source_label: str, source_key: Dict[str, Any], target_label: str, target_key: Dict[str, Any], client: Any) -> OperationResult async

Delete a relationship from the graph.

Parameters:

Name Type Description Default
rel_type str

Type of relationship to delete.

required
source_label str

Label of source node.

required
source_key Dict[str, Any]

Properties to identify source node.

required
target_label str

Label of target node.

required
target_key Dict[str, Any]

Properties to identify target node.

required
client Any

Neo4j client.

required

Returns:

Type Description
OperationResult

OperationResult with operation details.

ensure_indexes(client: Any) -> bool async

Ensure required indexes exist.

Creates both regular and vector indexes for memory nodes.

Parameters:

Name Type Description Default
client Any

Neo4j client.

required

Returns:

Type Description
bool

True if indexes were created/verified successfully.

MemoryUpdateController

Controller for orchestrating memory update operations.

This controller: 1. Extracts entities from content (using EntityExtractor) 2. Discovers relationships between entities (using RelationshipDiscoverer) 3. Decides operation type for each entity (using LLMOperationDecider) 4. Executes graph operations (using GraphOperations)

__init__(domain_config: Optional['DomainConfig'] = None, use_llm_extraction: bool = True, generate_embeddings: bool = True, similarity_threshold: float = 0.7, use_llm_decision: bool = False)

Initialize the update controller.

Parameters:

Name Type Description Default
domain_config Optional['DomainConfig']

Domain configuration defining entity types.

None
use_llm_extraction bool

Whether to use LLM for semantic extraction.

True
generate_embeddings bool

Whether to generate embeddings for entities.

True
similarity_threshold float

Threshold for similarity-based relationships.

0.7
use_llm_decision bool

Whether to use LLM for operation type decisions.

False

delete_entity(label: str, match_key: Dict[str, Any], client: Any) -> OperationResult async

Delete an entity from the memory graph.

Parameters:

Name Type Description Default
label str

Node label (e.g., "Question", "Topic").

required
match_key Dict[str, Any]

Properties to identify the node.

required
client Any

Neo4j client.

required

Returns:

Type Description
OperationResult

OperationResult with operation details.

delete_relationship(rel_type: str, source_label: str, source_key: Dict[str, Any], target_label: str, target_key: Dict[str, Any], client: Any) -> OperationResult async

Delete a relationship from the memory graph.

Parameters:

Name Type Description Default
rel_type str

Relationship type.

required
source_label str

Label of source node.

required
source_key Dict[str, Any]

Properties to identify source node.

required
target_label str

Label of target node.

required
target_key Dict[str, Any]

Properties to identify target node.

required
client Any

Neo4j client.

required

Returns:

Type Description
OperationResult

OperationResult with operation details.

Create a relationship between two existing entities.

Parameters:

Name Type Description Default
source_label str

Label of source node.

required
source_key Dict[str, Any]

Properties to identify source node.

required
target_label str

Label of target node.

required
target_key Dict[str, Any]

Properties to identify target node.

required
relationship_type str

Type of relationship to create.

required
client Any

Neo4j client.

required
properties Optional[Dict[str, Any]]

Optional relationship properties.

None

Returns:

Type Description
OperationResult

OperationResult for the relationship creation.

store_knowledge(content: str, content_type: str = 'text', client: Any = None, aigise_session_id: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None) -> UpdateResult async

Store knowledge in the memory graph.

Generic method for storing any type of content.

Parameters:

Name Type Description Default
content str

Content to store.

required
content_type str

Type of content ('text', 'code', 'question', 'answer').

'text'
client Any

Neo4j client.

None
aigise_session_id Optional[str]

Optional session ID.

None
metadata Optional[Dict[str, Any]]

Additional metadata.

None

Returns:

Type Description
UpdateResult

UpdateResult with operation details.

store_knowledge_with_decision(content: str, content_type: str = 'text', client: Any = None, aigise_session_id: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None) -> UpdateResult async

Store knowledge using LLM to decide operation type for each entity.

This method uses the LLMOperationDecider to intelligently decide whether to ADD, UPDATE, DELETE, or skip each extracted entity based on what already exists in the graph.

Parameters:

Name Type Description Default
content str

Content to store.

required
content_type str

Type of content.

'text'
client Any

Neo4j client.

None
aigise_session_id Optional[str]

Optional session ID.

None
metadata Optional[Dict[str, Any]]

Additional metadata.

None

Returns:

Type Description
UpdateResult

UpdateResult with operation details.

store_qa_pair(question: str, answer: str, answering_agent: str, answering_model: str, client: Any, aigise_session_id: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None) -> UpdateResult async

Store a question-answer pair in the memory graph.

This is the main entry point for storing Q&A knowledge. It extracts entities, discovers relationships, and persists to Neo4j.

Parameters:

Name Type Description Default
question str

The question text.

required
answer str

The answer text.

required
answering_agent str

Name of the agent that generated the answer.

required
answering_model str

Model used to generate the answer.

required
client Any

Neo4j client.

required
aigise_session_id Optional[str]

Optional session ID for tracking.

None
metadata Optional[Dict[str, Any]]

Additional metadata to store.

None

Returns:

Type Description
UpdateResult

UpdateResult with operation details.

_get_update_controller() -> MemoryUpdateController

Get or create the update controller singleton.

cache_qa_pair(question: str, answer: str, answering_agent: str, answering_model: str, *, metadata: Optional[Dict[str, Any]] = None, tool_context: ToolContext) -> Dict[str, Any] async

Cache a question-answer pair in the memory graph.

Use this tool AFTER successfully answering a question to store it for future retrieval. This enables the agent to quickly answer similar questions in the future.

The system automatically: - Generates embeddings for similarity search - Extracts topics and code references - Creates relationships to related entities - Finds connections to existing cached questions

Parameters:

Name Type Description Default
question str

The question text that was asked.

required
answer str

The answer text that was generated.

required
answering_agent str

Name of the agent that generated the answer.

required
answering_model str

Model identifier used to generate the answer.

required
metadata Optional[Dict[str, Any]]

Optional additional metadata to store.

None

Returns:

Type Description
Dict[str, Any]

Dictionary with:

Dict[str, Any]
  • success: True if caching succeeded
Dict[str, Any]
  • question_hash: Hash of the question for lookup
Dict[str, Any]
  • answer_id: Unique ID of the stored answer
Dict[str, Any]
  • entities_added: Number of entities created
Dict[str, Any]
  • relationships_added: Number of relationships created
Dict[str, Any]
  • topics_identified: Topics extracted from the Q&A
Dict[str, Any]
  • related_questions: Number of similar questions found

delete_from_memory(node_label: str, node_key: Dict[str, Any], *, tool_context: ToolContext) -> Dict[str, Any] async

Delete an entity from the memory graph.

Use this to remove outdated or incorrect cached knowledge. This will also remove all relationships connected to the node.

Parameters:

Name Type Description Default
node_label str

Label of the node to delete. Options: - "Question": A cached question - "Answer": A cached answer - "Topic": A topic/concept - "Function": A function entity - "Class": A class entity - "File": A file entity

required
node_key Dict[str, Any]

Properties to identify the node. Examples: - {"question_hash": "..."} for Question - {"name": "..."} for Topic - {"name": "...", "file_path": "..."} for Function/Class - {"path": "..."} for File

required

Returns:

Type Description
Dict[str, Any]

Dictionary with:

Dict[str, Any]
  • success: True if entity was deleted
Dict[str, Any]
  • message: Description of what was deleted
Dict[str, Any]
  • error: Error message if deletion failed

delete_relationship_from_memory(relationship_type: str, source_label: str, source_key: Dict[str, Any], target_label: str, target_key: Dict[str, Any], *, tool_context: ToolContext) -> Dict[str, Any] async

Delete a relationship from the memory graph.

Use this to remove incorrect or outdated connections between entities.

Parameters:

Name Type Description Default
relationship_type str

Type of relationship to delete. Options: - "ABOUT": What something is about - "HAS_ANSWER": Question to answer link - "HAS_TOPIC": Content to topic link - "RELATED_TO": General relation - "MENTIONS": Content mentions code entity

required
source_label str

Label of the source node.

required
source_key Dict[str, Any]

Properties to identify the source node.

required
target_label str

Label of the target node.

required
target_key Dict[str, Any]

Properties to identify the target node.

required

Returns:

Type Description
Dict[str, Any]

Dictionary with:

Dict[str, Any]
  • success: True if relationship was deleted
Dict[str, Any]
  • message: Description of what was deleted

ensure_memory_indexes(*, tool_context: ToolContext) -> Dict[str, Any] async

Ensure memory graph indexes are created.

This tool creates the necessary indexes for efficient querying: - Regular indexes for exact match lookups - Vector indexes for similarity search

You typically don't need to call this manually as it's called automatically during the first cache operation.

Returns:

Type Description
Dict[str, Any]

Dictionary with:

Dict[str, Any]
  • success: True if indexes were created/verified
Dict[str, Any]
  • message: Description of indexes created

get_merged_domain(*domain_names: str) -> DomainConfig

Get a merged domain from multiple registered domains.

Parameters:

Name Type Description Default
*domain_names str

Names of domains to merge.

()

Returns:

Type Description
DomainConfig

Merged domain configuration.

Raises:

Type Description
ValueError

If any domain name is not found.

Create a relationship between two entities in the memory graph.

Use this to manually connect related knowledge that the automatic extraction might have missed.

Parameters:

Name Type Description Default
source_label str

Label of the source node (e.g., "Question").

required
source_key Dict[str, Any]

Properties to identify source (e.g., {"question_hash": "..."}).

required
target_label str

Label of the target node (e.g., "Topic").

required
target_key Dict[str, Any]

Properties to identify target (e.g., {"name": "..."}).

required
relationship_type str

Type of relationship. Options: - "ABOUT": What something is about - "HAS_ANSWER": Question to answer link - "HAS_TOPIC": Content to topic link - "RELATED_TO": General relation - "MENTIONS": Content mentions code entity

required
properties Optional[Dict[str, Any]]

Optional properties for the relationship.

None

Returns:

Type Description
Dict[str, Any]

Dictionary with:

Dict[str, Any]
  • success: True if relationship was created
Dict[str, Any]
  • relationship_type: The type of relationship created
Dict[str, Any]
  • message: Description of what was linked

requires_sandbox(*sandbox_types: str) -> Callable[[F], F]

Universal decorator for declaring sandbox dependencies.

This decorator works for both: - Tool functions: Only marks the function with metadata - Toolset factories: Marks function AND injects metadata into returned instance

The decorator is purely declarative - it does not create or fetch sandboxes. It only adds __sandbox_requirements__ metadata for static analysis via collect_sandbox_dependencies().

Parameters:

Name Type Description Default
*sandbox_types str

Variable number of sandbox type names that the tool or toolset depends on (e.g., "main", "gdb_mcp", "neo4j").

()

Returns:

Type Description
Callable[[F], F]

A decorator function that adds sandbox_requirements metadata.

safe_tool_execution(func: F) -> F

Decorator to wrap tool functions with error handling.

Catches all exceptions and returns a formatted error message with backtrace. Works for both sync and async functions.

Returns:

Type Description
F

dict with "error" key containing failure message and backtrace

store_knowledge(content: str, content_type: str = 'text', *, metadata: Optional[Dict[str, Any]] = None, tool_context: ToolContext) -> Dict[str, Any] async

Store knowledge in the memory graph.

Use this tool to persist any kind of knowledge for future retrieval. The system will automatically extract entities, topics, and relationships from the content.

Parameters:

Name Type Description Default
content str

The content to store.

required
content_type str

Type of content. Options: - "text": Generic text content - "code": Code content (will extract functions/classes) - "question": A question to cache - "answer": An answer to cache

'text'
metadata Optional[Dict[str, Any]]

Optional additional metadata to store.

None

Returns:

Type Description
Dict[str, Any]

Dictionary with:

Dict[str, Any]
  • success: True if storage succeeded
Dict[str, Any]
  • entities_added: Number of entities created
Dict[str, Any]
  • entities_updated: Number of existing entities updated
Dict[str, Any]
  • relationships_added: Number of relationships created
Dict[str, Any]
  • message: Summary of what was stored