Merge pull request #22356 from BerriAI/litellm_key_list_filters

[Feature] Key list endpoint: Add project_id and access_group_id filters

litellm/proxy/management_endpoints/key_management_endpoints.py

@@ -3983,6 +3983,10 @@ async def list_keys(
     status: Optional[str] = Query(
         None, description="Filter by status (e.g. 'deleted')"
     ),
+    project_id: Optional[str] = Query(None, description="Filter keys by project ID"),
+    access_group_id: Optional[str] = Query(
+        None, description="Filter keys by access group ID"
+    ),
 ) -> KeyListResponseObject:
     """
     List all keys for a given user / team / organization.
@@ -4076,6 +4080,8 @@ async def list_keys(
             sort_order=sort_order,
             expand=expand,
             status=status,
+            project_id=project_id,
+            access_group_id=access_group_id,
         )
 
         verbose_proxy_logger.debug("Successfully prepared response")
@@ -4252,6 +4258,8 @@ def _build_key_filter_conditions(
     admin_team_ids: Optional[List[str]],
     member_team_ids: Optional[List[str]] = None,
     include_created_by_keys: bool = False,
+    project_id: Optional[str] = None,
+    access_group_id: Optional[str] = None,
 ) -> Dict[str, Union[str, Dict[str, Any], List[Dict[str, Any]]]]:
     """Build filter conditions for key listing.
 
@@ -4343,6 +4351,13 @@ def _build_key_filter_conditions(
     elif len(or_conditions) == 1:
         where.update(or_conditions[0])
 
+    # Apply project_id and access_group_id as global AND filters so they
+    # narrow results across all visibility conditions (own keys, team keys, etc.)
+    if project_id:
+        where = {"AND": [where, {"project_id": project_id}]}
+    if access_group_id:
+        where = {"AND": [where, {"access_group_ids": {"hasSome": [access_group_id]}}]}
+
     verbose_proxy_logger.debug(f"Filter conditions: {where}")
     return where
 
@@ -4369,6 +4384,8 @@ async def _list_key_helper(
     sort_order: str = "desc",
     expand: Optional[List[str]] = None,
     status: Optional[str] = None,
+    project_id: Optional[str] = None,
+    access_group_id: Optional[str] = None,
 ) -> KeyListResponseObject:
     """
     Helper function to list keys
@@ -4402,6 +4419,8 @@ async def _list_key_helper(
         admin_team_ids=admin_team_ids,
         member_team_ids=member_team_ids,
         include_created_by_keys=include_created_by_keys,
+        project_id=project_id,
+        access_group_id=access_group_id,
     )
 
     # Calculate skip for pagination

tests/test_litellm/proxy/management_endpoints/test_key_management_endpoints.py

@@ -6070,6 +6070,121 @@ async def test_build_key_filter_admin_all_member_overlap():
                 )
 
 
+@pytest.mark.asyncio
+async def test_build_key_filter_project_id():
+    """
+    Test that project_id is applied as a global AND condition, narrowing all results
+    to keys that belong to the specified project.
+    """
+    from litellm.proxy.management_endpoints.key_management_endpoints import (
+        _build_key_filter_conditions,
+    )
+
+    user_id = "user-123"
+    project_id = "proj-abc"
+
+    where = _build_key_filter_conditions(
+        user_id=user_id,
+        team_id=None,
+        organization_id=None,
+        key_alias=None,
+        key_hash=None,
+        exclude_team_id=None,
+        admin_team_ids=None,
+        member_team_ids=None,
+        include_created_by_keys=False,
+        project_id=project_id,
+    )
+
+    # Should be wrapped in a top-level AND for the project_id filter
+    assert "AND" in where
+    and_parts = where["AND"]
+    assert len(and_parts) == 2
+
+    # Second part of AND should be the project_id filter
+    assert {"project_id": project_id} in and_parts
+
+
+@pytest.mark.asyncio
+async def test_build_key_filter_access_group_id():
+    """
+    Test that access_group_id is applied as a global AND condition using hasSome,
+    narrowing results to keys whose access_group_ids array contains the given ID.
+    """
+    from litellm.proxy.management_endpoints.key_management_endpoints import (
+        _build_key_filter_conditions,
+    )
+
+    user_id = "user-123"
+    access_group_id = "ag-xyz"
+
+    where = _build_key_filter_conditions(
+        user_id=user_id,
+        team_id=None,
+        organization_id=None,
+        key_alias=None,
+        key_hash=None,
+        exclude_team_id=None,
+        admin_team_ids=None,
+        member_team_ids=None,
+        include_created_by_keys=False,
+        access_group_id=access_group_id,
+    )
+
+    # Should be wrapped in a top-level AND for the access_group_id filter
+    assert "AND" in where
+    and_parts = where["AND"]
+    assert len(and_parts) == 2
+
+    # Second part of AND should use hasSome for the array field
+    assert {"access_group_ids": {"hasSome": [access_group_id]}} in and_parts
+
+
+@pytest.mark.asyncio
+async def test_build_key_filter_project_id_and_access_group_id():
+    """
+    Test that project_id and access_group_id stack correctly when both are provided.
+    Both should be applied as AND conditions, narrowing results to keys that match both.
+    """
+    from litellm.proxy.management_endpoints.key_management_endpoints import (
+        _build_key_filter_conditions,
+    )
+
+    user_id = "user-123"
+    project_id = "proj-abc"
+    access_group_id = "ag-xyz"
+
+    where = _build_key_filter_conditions(
+        user_id=user_id,
+        team_id=None,
+        organization_id=None,
+        key_alias=None,
+        key_hash=None,
+        exclude_team_id=None,
+        admin_team_ids=None,
+        member_team_ids=None,
+        include_created_by_keys=False,
+        project_id=project_id,
+        access_group_id=access_group_id,
+    )
+
+    # After project_id: {"AND": [visibility_where, {"project_id": ...}]}
+    # After access_group_id: {"AND": [above, {"access_group_ids": ...}]}
+    assert "AND" in where
+    outer_and = where["AND"]
+    assert len(outer_and) == 2
+
+    # The access_group_ids filter is the outermost AND
+    access_group_filter = outer_and[1]
+    assert access_group_filter == {"access_group_ids": {"hasSome": [access_group_id]}}
+
+    # The project_id filter is nested one level in
+    inner = outer_and[0]
+    assert "AND" in inner
+    inner_and = inner["AND"]
+    assert {"project_id": project_id} in inner_and
+
+
 @pytest.mark.asyncio
 async def test_get_member_team_ids():
     """