FEAT: add update delete tool

Author: iwakitakuma33

src/mcp_server_qdrant/mcp_server.py

@@ -59,7 +59,8 @@ def format_entry(self, entry: Entry) -> str:
         Feel free to override this method in your subclass to customize the format of the entry.
         """
         entry_metadata = json.dumps(entry.metadata) if entry.metadata else ""
-        return f"<entry><content>{entry.content}</content><metadata>{entry_metadata}</metadata></entry>"
+        entry_id = f"<id>{entry.id}</id>" if entry.id else ""
+        return f"<entry>{entry_id}<content>{entry.content}</content><metadata>{entry_metadata}</metadata></entry>"
 
     def setup_tools(self):
         """
@@ -68,6 +69,10 @@ def setup_tools(self):
 
         async def store(
             ctx: Context,
+            id: Annotated[
+                str | None,
+                Field(description="Point ID. If omitted, a new point is created."),
+            ],
             information: Annotated[str, Field(description="Text to store")],
             collection_name: Annotated[
                 str, Field(description="The collection to store the information in")
@@ -93,7 +98,7 @@ async def store(
             """
             await ctx.debug(f"Storing information {information} in Qdrant")
 
-            entry = Entry(content=information, metadata=metadata)
+            entry = Entry(content=information, metadata=metadata, id=id)
 
             await self.qdrant_connector.store(entry, collection_name=collection_name)
             if collection_name:
@@ -160,16 +165,147 @@ async def find(
                 store_foo, {"collection_name": self.qdrant_settings.collection_name}
             )
 
+        # Add new tools for point operations
+        async def get_point(
+            ctx: Context,
+            point_id: Annotated[
+                str, Field(description="The ID of the point to retrieve")
+            ],
+            collection_name: Annotated[
+                str, Field(description="The collection to get the point from")
+            ],
+        ) -> str:
+            """
+            Get a specific point by its ID.
+            :param ctx: The context for the request.
+            :param point_id: The ID of the point to retrieve.
+            :param collection_name: The name of the collection to get the point from.
+            :return: The point information or error message.
+            """
+            await ctx.debug(
+                f"Getting point {point_id} from collection {collection_name}"
+            )
+
+            entry = await self.qdrant_connector.get_point_by_id(
+                point_id, collection_name=collection_name
+            )
+
+            if entry:
+                return self.format_entry(entry)
+            else:
+                return f"Point with ID {point_id} not found in collection {collection_name}"
+
+        async def delete_point(
+            ctx: Context,
+            point_id: Annotated[
+                str, Field(description="The ID of the point to delete")
+            ],
+            collection_name: Annotated[
+                str, Field(description="The collection to delete the point from")
+            ],
+        ) -> str:
+            """
+            Delete a specific point by its ID.
+            :param ctx: The context for the request.
+            :param point_id: The ID of the point to delete.
+            :param collection_name: The name of the collection to delete the point from.
+            :return: Success or error message.
+            """
+            await ctx.debug(
+                f"Deleting point {point_id} from collection {collection_name}"
+            )
+
+            success = await self.qdrant_connector.delete_point_by_id(
+                point_id, collection_name=collection_name
+            )
+
+            if success:
+                return f"Successfully deleted point {point_id} from collection {collection_name}"
+            else:
+                return f"Failed to delete point {point_id} - point not found or collection doesn't exist"
+
+        async def update_point_payload(
+            ctx: Context,
+            point_id: Annotated[
+                str, Field(description="The ID of the point to update")
+            ],
+            collection_name: Annotated[
+                str, Field(description="The collection containing the point")
+            ],
+            metadata: Annotated[
+                Metadata,
+                Field(
+                    description="New metadata to set for the point. Any json is accepted."
+                ),
+            ],
+        ) -> str:
+            """
+            Update the payload (metadata) of a specific point by its ID.
+            :param ctx: The context for the request.
+            :param point_id: The ID of the point to update.
+            :param collection_name: The name of the collection containing the point.
+            :param metadata: New metadata to set for the point.
+            :return: Success or error message.
+            """
+            await ctx.debug(
+                f"Updating payload for point {point_id} in collection {collection_name}"
+            )
+
+            success = await self.qdrant_connector.update_point_payload(
+                point_id, metadata, collection_name=collection_name
+            )
+
+            if success:
+                return f"Successfully updated payload for point {point_id} in collection {collection_name}"
+            else:
+                return f"Failed to update payload for point {point_id} - point not found or collection doesn't exist"
+
+        # Apply collection name defaults to new tools
+        get_point_foo = get_point
+        delete_point_foo = delete_point
+        update_point_payload_foo = update_point_payload
+
+        if self.qdrant_settings.collection_name:
+            get_point_foo = make_partial_function(
+                get_point_foo, {"collection_name": self.qdrant_settings.collection_name}
+            )
+            delete_point_foo = make_partial_function(
+                delete_point_foo,
+                {"collection_name": self.qdrant_settings.collection_name},
+            )
+            update_point_payload_foo = make_partial_function(
+                update_point_payload_foo,
+                {"collection_name": self.qdrant_settings.collection_name},
+            )
+
         self.tool(
             find_foo,
             name="qdrant-find",
             description=self.tool_settings.tool_find_description,
         )
 
+        self.tool(
+            get_point_foo,
+            name="qdrant-get-point",
+            description="Get a specific point by its ID from a Qdrant collection.",
+        )
+
         if not self.qdrant_settings.read_only:
             # Those methods can modify the database
             self.tool(
                 store_foo,
                 name="qdrant-store",
                 description=self.tool_settings.tool_store_description,
             )
+
+            self.tool(
+                delete_point_foo,
+                name="qdrant-delete-point",
+                description="Delete a specific point by its ID from a Qdrant collection.",
+            )
+
+            self.tool(
+                update_point_payload_foo,
+                name="qdrant-update-point-payload",
+                description="Update the payload (metadata) of a specific point by its ID.",
+            )

src/mcp_server_qdrant/qdrant.py

@@ -19,6 +19,7 @@ class Entry(BaseModel):
     A single entry in the Qdrant collection.
     """
 
+    id: str | None = None
     content: str
     metadata: Metadata | None = None
 
@@ -76,14 +77,15 @@ async def store(self, entry: Entry, *, collection_name: str | None = None):
         # it should unlock usage of server-side inference.
         embeddings = await self._embedding_provider.embed_documents([entry.content])
 
+        point_id = uuid.uuid4().hex if entry.id is None else entry.id
         # Add to Qdrant
         vector_name = self._embedding_provider.get_vector_name()
         payload = {"document": entry.content, METADATA_PATH: entry.metadata}
         await self._client.upsert(
             collection_name=collection_name,
             points=[
                 models.PointStruct(
-                    id=uuid.uuid4().hex,
+                    id=point_id,
                     vector={vector_name: embeddings[0]},
                     payload=payload,
                 )
@@ -131,6 +133,7 @@ async def search(
 
         return [
             Entry(
+                id=str(result.id),
                 content=result.payload["document"],
                 metadata=result.payload.get("metadata"),
             )
@@ -168,3 +171,109 @@ async def _ensure_collection_exists(self, collection_name: str):
                         field_name=field_name,
                         field_schema=field_type,
                     )
+
+    async def get_point_by_id(
+        self,
+        point_id: str,
+        *,
+        collection_name: str | None = None,
+    ) -> Entry | None:
+        """
+        Retrieve a specific point by its ID.
+        :param point_id: The ID of the point to retrieve.
+        :param collection_name: The name of the collection, optional.
+        :return: The entry if found, None otherwise.
+        """
+        collection_name = collection_name or self._default_collection_name
+        assert collection_name is not None
+
+        point = await self._client.retrieve(
+            collection_name=collection_name,
+            ids=[point_id],
+            with_payload=True,
+            with_vectors=False,
+        )
+
+        if point and len(point) > 0:
+            result = point[0]
+            return Entry(
+                id=str(result.id),
+                content=result.payload["document"],
+                metadata=result.payload.get("metadata"),
+            )
+        return None
+
+    async def delete_point_by_id(
+        self,
+        point_id: str,
+        *,
+        collection_name: str | None = None,
+    ) -> bool:
+        """
+        Delete a specific point by its ID.
+        :param point_id: The ID of the point to delete.
+        :param collection_name: The name of the collection, optional.
+        :return: True if the point was deleted, False if not found.
+        """
+        collection_name = collection_name or self._default_collection_name
+        assert collection_name is not None
+
+        collection_exists = await self._client.collection_exists(collection_name)
+        if not collection_exists:
+            return False
+
+        # Check if point exists before trying to delete
+        existing_point = await self.get_point_by_id(
+            point_id, collection_name=collection_name
+        )
+        if existing_point is None:
+            return False
+
+        try:
+            await self._client.delete(
+                collection_name=collection_name,
+                points_selector=models.PointIdsList(points=[point_id]),
+            )
+            return True
+        except Exception as e:
+            logger.error(f"Failed to delete point {point_id}: {e}")
+            return False
+
+    async def update_point_payload(
+        self,
+        point_id: str,
+        metadata: Metadata,
+        *,
+        collection_name: str | None = None,
+    ) -> bool:
+        """
+        Update the payload (metadata) of a specific point by its ID.
+        :param point_id: The ID of the point to update.
+        :param metadata: New metadata to set for the point.
+        :param collection_name: The name of the collection, optional.
+        :return: True if the point was updated, False if not found.
+        """
+        collection_name = collection_name or self._default_collection_name
+        assert collection_name is not None
+
+        collection_exists = await self._client.collection_exists(collection_name)
+        if not collection_exists:
+            return False
+
+        # Check if point exists before trying to update
+        existing_point = await self.get_point_by_id(
+            point_id, collection_name=collection_name
+        )
+        if existing_point is None:
+            return False
+
+        try:
+            await self._client.set_payload(
+                collection_name=collection_name,
+                payload={METADATA_PATH: metadata},
+                points=[point_id],
+            )
+            return True
+        except Exception as e:
+            logger.error(f"Failed to update payload for point {point_id}: {e}")
+            return False