feat: add the ability to find/delete by deployment id [PRO-1150] (#4421)

* feat: add the ability to find/delete by deployment id

* fix: add ability to delete by deployment id

---------

Co-authored-by: Shubham Naik <shub@memgpt.ai>

letta/server/rest_api/routers/v1/internal_templates.py

@@ -1,6 +1,7 @@
-from typing import Optional
+from typing import List, Optional
 
-from fastapi import APIRouter, Body, Depends, Header, HTTPException
+from fastapi import APIRouter, Body, Depends, Header, HTTPException, Query
+from pydantic import BaseModel
 
 from letta.schemas.agent import AgentState, InternalTemplateAgentCreate
 from letta.schemas.block import Block, InternalTemplateBlockCreate
@@ -57,3 +58,216 @@ async def create_block(
         return await server.block_manager.create_or_update_block_async(block, actor=actor)
     except Exception as e:
         raise HTTPException(status_code=500, detail=str(e))
+
+
+class DeploymentEntity(BaseModel):
+    """A deployment entity."""
+
+    id: str
+    type: str
+    name: Optional[str] = None
+    description: Optional[str] = None
+
+
+class ListDeploymentEntitiesResponse(BaseModel):
+    """Response model for listing deployment entities."""
+
+    entities: List[DeploymentEntity] = []
+    total_count: int
+    deployment_id: str
+    message: str
+
+
+class DeleteDeploymentResponse(BaseModel):
+    """Response model for delete deployment operation."""
+
+    deleted_blocks: List[str] = []
+    deleted_agents: List[str] = []
+    deleted_groups: List[str] = []
+    message: str
+
+
+@router.get("/deployment/{deployment_id}", response_model=ListDeploymentEntitiesResponse, operation_id="list_deployment_entities")
+async def list_deployment_entities(
+    deployment_id: str,
+    server: "SyncServer" = Depends(get_letta_server),
+    actor_id: Optional[str] = Header(None, alias="user_id"),
+    entity_types: Optional[List[str]] = Query(None, description="Filter by entity types (block, agent, group)"),
+):
+    """
+    List all entities (blocks, agents, groups) with the specified deployment_id.
+    Optionally filter by entity types.
+    """
+    try:
+        actor = await server.user_manager.get_actor_or_default_async(actor_id=actor_id)
+
+        entities = []
+
+        # Parse entity_types filter - support both array and comma-separated string
+        allowed_types = {"block", "agent", "group"}
+        if entity_types is None:
+            # If no filter specified, include all types
+            types_to_include = allowed_types
+        else:
+            # Handle comma-separated strings in a single item
+            if len(entity_types) == 1 and "," in entity_types[0]:
+                entity_types = [t.strip() for t in entity_types[0].split(",")]
+
+            # Validate and filter types
+            types_to_include = {t.lower() for t in entity_types if t.lower() in allowed_types}
+            if not types_to_include:
+                types_to_include = allowed_types  # Default to all if invalid types provided
+
+        # Query blocks if requested
+        if "block" in types_to_include:
+            from sqlalchemy import select
+
+            from letta.orm.block import Block as BlockModel
+            from letta.server.db import db_registry
+
+            async with db_registry.async_session() as session:
+                block_query = select(BlockModel).where(
+                    BlockModel.deployment_id == deployment_id, BlockModel.organization_id == actor.organization_id
+                )
+                result = await session.execute(block_query)
+                blocks = result.scalars().all()
+
+                for block in blocks:
+                    entities.append(
+                        DeploymentEntity(
+                            id=block.id,
+                            type="block",
+                            name=getattr(block, "template_name", None) or getattr(block, "label", None),
+                            description=block.description,
+                        )
+                    )
+
+        # Query agents if requested
+        if "agent" in types_to_include:
+            from letta.orm.agent import Agent as AgentModel
+
+            async with db_registry.async_session() as session:
+                agent_query = select(AgentModel).where(
+                    AgentModel.deployment_id == deployment_id, AgentModel.organization_id == actor.organization_id
+                )
+                result = await session.execute(agent_query)
+                agents = result.scalars().all()
+
+                for agent in agents:
+                    entities.append(DeploymentEntity(id=agent.id, type="agent", name=agent.name, description=agent.description))
+
+        # Query groups if requested
+        if "group" in types_to_include:
+            from letta.orm.group import Group as GroupModel
+
+            async with db_registry.async_session() as session:
+                group_query = select(GroupModel).where(
+                    GroupModel.deployment_id == deployment_id, GroupModel.organization_id == actor.organization_id
+                )
+                result = await session.execute(group_query)
+                groups = result.scalars().all()
+
+                for group in groups:
+                    entities.append(
+                        DeploymentEntity(
+                            id=group.id,
+                            type="group",
+                            name=None,  # Groups don't have a name field
+                            description=group.description,
+                        )
+                    )
+
+        message = f"Found {len(entities)} entities for deployment {deployment_id}"
+        if entity_types:
+            message += f" (filtered by types: {', '.join(types_to_include)})"
+
+        return ListDeploymentEntitiesResponse(entities=entities, total_count=len(entities), deployment_id=deployment_id, message=message)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+@router.delete("/deployment/{deployment_id}", response_model=DeleteDeploymentResponse, operation_id="delete_deployment")
+async def delete_deployment(
+    deployment_id: str,
+    server: "SyncServer" = Depends(get_letta_server),
+    actor_id: Optional[str] = Header(None, alias="user_id"),
+):
+    """
+    Delete all entities (blocks, agents, groups) with the specified deployment_id.
+    Deletion order: blocks -> agents -> groups to maintain referential integrity.
+    """
+    try:
+        actor = await server.user_manager.get_actor_or_default_async(actor_id=actor_id)
+
+        deleted_blocks = []
+        deleted_agents = []
+        deleted_groups = []
+
+        # First delete blocks
+        from sqlalchemy import select
+
+        from letta.orm.block import Block as BlockModel
+        from letta.server.db import db_registry
+
+        async with db_registry.async_session() as session:
+            # Get all blocks with the deployment_id
+            block_query = select(BlockModel).where(
+                BlockModel.deployment_id == deployment_id, BlockModel.organization_id == actor.organization_id
+            )
+            result = await session.execute(block_query)
+            blocks = result.scalars().all()
+
+            for block in blocks:
+                try:
+                    await server.block_manager.delete_block_async(block.id, actor)
+                    deleted_blocks.append(block.id)
+                except Exception as e:
+                    # Continue deleting other blocks even if one fails
+                    print(f"Failed to delete block {block.id}: {e}")
+
+        # Then delete agents
+        from letta.orm.agent import Agent as AgentModel
+
+        async with db_registry.async_session() as session:
+            # Get all agents with the deployment_id
+            agent_query = select(AgentModel).where(
+                AgentModel.deployment_id == deployment_id, AgentModel.organization_id == actor.organization_id
+            )
+            result = await session.execute(agent_query)
+            agents = result.scalars().all()
+
+            for agent in agents:
+                try:
+                    await server.agent_manager.delete_agent_async(agent.id, actor)
+                    deleted_agents.append(agent.id)
+                except Exception as e:
+                    # Continue deleting other agents even if one fails
+                    print(f"Failed to delete agent {agent.id}: {e}")
+
+        # Finally delete groups
+        from letta.orm.group import Group as GroupModel
+
+        async with db_registry.async_session() as session:
+            # Get all groups with the deployment_id
+            group_query = select(GroupModel).where(
+                GroupModel.deployment_id == deployment_id, GroupModel.organization_id == actor.organization_id
+            )
+            result = await session.execute(group_query)
+            groups = result.scalars().all()
+
+            for group in groups:
+                try:
+                    await server.group_manager.delete_group_async(group.id, actor)
+                    deleted_groups.append(group.id)
+                except Exception as e:
+                    # Continue deleting other groups even if one fails
+                    print(f"Failed to delete group {group.id}: {e}")
+
+        total_deleted = len(deleted_blocks) + len(deleted_agents) + len(deleted_groups)
+        message = f"Successfully deleted {total_deleted} entities from deployment {deployment_id}"
+
+        return DeleteDeploymentResponse(
+            deleted_blocks=deleted_blocks, deleted_agents=deleted_agents, deleted_groups=deleted_groups, message=message
+        )
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))