search_memory_nodes
Find relevant node summaries in graph memory by searching with queries, filtering by groups or entity types, and centering around specific nodes.
Instructions
Search the graph memory for relevant node summaries.
These contain a summary of all of a node's relationships with other nodes.
Note: entity is a single entity type to filter results (permitted: "Preference", "Procedure").
Args:
query: The search query
group_ids: Optional list of group IDs to filter results
max_nodes: Maximum number of nodes to return (default: 10)
center_node_uuid: Optional UUID of a node to center the search around
entity: Optional single entity type to filter results (permitted: "Preference", "Procedure")
Returns:
List of node dictionaries containing search results
Example:
search_memory_nodes(
query="project architecture decisions",
group_ids=["knowledge-smith"],
max_nodes=5
)
@REQ: REQ-graphiti-chunk-mcp
@BP: BP-graphiti-chunk-mcp
@TASK: TASK-007-MCPTools
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| query | Yes | ||
| group_ids | No | ||
| max_nodes | No | ||
| center_node_uuid | No | ||
| entity | No |
Implementation Reference
- rbt_mcp_server/server.py:25-67 (handler)MCP tool handler for 'search_memory_nodes'. Includes @mcp.tool() decorator (registration), type hints and docstring (schema), and delegates to helper implementation.@mcp.tool() async def search_memory_nodes( query: str, group_ids: Optional[List[str]] = None, max_nodes: int = 10, center_node_uuid: Optional[str] = None, entity: str = "", ) -> List[Dict[str, Any]]: """ Search the graph memory for relevant node summaries. These contain a summary of all of a node's relationships with other nodes. Note: entity is a single entity type to filter results (permitted: "Preference", "Procedure"). Args: query: The search query group_ids: Optional list of group IDs to filter results max_nodes: Maximum number of nodes to return (default: 10) center_node_uuid: Optional UUID of a node to center the search around entity: Optional single entity type to filter results (permitted: "Preference", "Procedure") Returns: List of node dictionaries containing search results Example: search_memory_nodes( query="project architecture decisions", group_ids=["knowledge-smith"], max_nodes=5 ) @REQ: REQ-graphiti-chunk-mcp @BP: BP-graphiti-chunk-mcp @TASK: TASK-007-MCPTools """ return await graphiti_tools.search_nodes_impl( query=query, group_ids=group_ids, max_nodes=max_nodes, center_node_uuid=center_node_uuid, entity=entity, )
- Core helper implementation performing the node search using GraphitiClient.search_nodes after creating the client.async def search_nodes_impl( query: str, group_ids: Optional[List[str]] = None, max_nodes: int = 10, center_node_uuid: Optional[str] = None, entity: str = "", ) -> List[Dict[str, Any]]: """ Search the graph memory for relevant node summaries. @REQ: REQ-graphiti-chunk-mcp @BP: BP-graphiti-chunk-mcp @TASK: TASK-007-MCPTools These contain a summary of all of a node's relationships with other nodes. Note: entity is a single entity type to filter results (permitted: "Preference", "Procedure"). Args: query: The search query group_ids: Optional list of group IDs to filter results max_nodes: Maximum number of nodes to return (default: 10) center_node_uuid: Optional UUID of a node to center the search around entity: Optional single entity type to filter results (permitted: "Preference", "Procedure") Returns: List of node dictionaries containing search results Raises: ToolError: If search operation fails """ try: client = get_graphiti_client() async with client: results = await client.search_nodes( query=query, center_node_uuid=center_node_uuid, group_ids=group_ids, max_nodes=max_nodes, ) # Filter by entity type if specified if entity: # Note: The entity filtering would need to be implemented # based on node labels or properties. For now, we return all results. pass return results except Exception as e: raise ToolError( "SEARCH_NODES_ERROR", f"Failed to search nodes: {str(e)}" ) from e
- Supporting utility to create GraphitiClient instance from environment variables, used by search_nodes_impl.def get_graphiti_client() -> GraphitiClient: """ Create and return a GraphitiClient instance using environment variables. @REQ: REQ-graphiti-chunk-mcp @BP: BP-graphiti-chunk-mcp @TASK: TASK-007-MCPTools Required environment variables: - NEO4J_URI: Neo4j database URI (e.g., 'bolt://localhost:7687') - NEO4J_USER: Neo4j username - NEO4J_PASSWORD: Neo4j password - OPENAI_API_KEY: OpenAI API key Returns: GraphitiClient instance Raises: ValueError: If required environment variables are missing """ neo4j_uri = os.environ.get("NEO4J_URI") neo4j_user = os.environ.get("NEO4J_USER") neo4j_password = os.environ.get("NEO4J_PASSWORD") openai_api_key = os.environ.get("OPENAI_API_KEY") if not all([neo4j_uri, neo4j_user, neo4j_password, openai_api_key]): raise ValueError( "Missing required environment variables. Please set: " "NEO4J_URI, NEO4J_USER, NEO4J_PASSWORD, OPENAI_API_KEY" ) return GraphitiClient( neo4j_uri=neo4j_uri, neo4j_user=neo4j_user, neo4j_password=neo4j_password, openai_api_key=openai_api_key, )
- rbt_mcp_server/server.py:25-25 (registration)The @mcp.tool() decorator registers the search_memory_nodes function as an MCP tool.@mcp.tool()
- rbt_mcp_server/server.py:27-32 (schema)Type hints defining the input schema for the tool.query: str, group_ids: Optional[List[str]] = None, max_nodes: int = 10, center_node_uuid: Optional[str] = None, entity: str = "", ) -> List[Dict[str, Any]]: