Knowledge Graph Memory Server

Instructions

Implementation Reference

• optimized_memory_mcp_server/operations/entity_ops.py:48-88 (handler)

  This is the core handler function for the 'add_observations' tool. It processes a batch of observations and updates the underlying database.

  async def add_observations(
      self, 
      observations: List[Dict[str, Any]], 
      batch_size: int = 1000
  ) -> Dict[str, List[str]]:
      """Add new observations to existing entities."""
      added_observations = {}
      
      async with self.pool.get_connection() as conn:
          async with self.pool.transaction(conn):
              for i in range(0, len(observations), batch_size):
                  batch = observations[i:i + batch_size]
                  
                  for obs in batch:
                      entity_name = sanitize_input(obs["entityName"])
                      new_contents = obs["contents"]
  
                      cursor = await conn.execute(
                          "SELECT observations FROM entities WHERE name = ?",
                          (entity_name,)
                      )
                      result = await cursor.fetchone()
                      if not result:
                          raise EntityNotFoundError(entity_name)
  
                      current_obs = result['observations'].split(',') if result['observations'] else []
                      current_obs.extend(new_contents)
                      
                      await conn.execute(
                          "UPDATE entities SET observations = ? WHERE name = ?",
                          (','.join(current_obs), entity_name)
                      )
                      added_observations[entity_name] = new_contents
  
      return added_observations
  
  async def delete_entities(
      self, 
      entity_names: List[str], 
      batch_size: int = 1000
  ) -> None:

• optimized_memory_mcp_server/main.py:154-173 (registration)

  The MCP tool registration for 'add_observations' in the main server file.

  types.Tool(
      name="add_observations",
      description="Add new observations to existing entities",
      inputSchema={
          "type": "object",
          "properties": {
              "observations": {
                  "type": "array",
                  "items": {
                      "type": "object",
                      "properties": {
                          "entityName": {"type": "string"},
                          "contents": {
                              "type": "array",
                              "items": {"type": "string"}
                          }
                      },
                      "required": ["entityName", "contents"],
                      "additionalProperties": False
                  }