fix(agents): cap a managed agent at the invoking team's agents

resolve_agent_access returned the managed policy's grants before the
agent_caller ceiling was applied, so a managed agent acting on behalf of a
user reached agents that user's team was never granted. Intersect with the
caller ceiling the unmanaged path already honours.
This commit is contained in:
Joshua Valluru 2026-09-29 16:00:02 -07:00
parent c41946a9d2
commit 2d2782e4a6
2 changed files with 26 additions and 1 deletions

View file

@ -650,7 +650,9 @@ async def _managed_actor_agent_access(auth: UserAPIKeyAuth) -> AgentAccess:
from litellm.proxy.agent_endpoints.auth.agent_access_groups import resolve_managed_agent_ceilings
ceilings: Final = await resolve_managed_agent_ceilings(agent)
capped: Final = frozenset(target for target in own if all(target in ceiling.agent_ids for ceiling in ceilings))
grouped: Final = frozenset(target for target in own if all(target in ceiling.agent_ids for ceiling in ceilings))
caller: Final = await AgentRequestHandler._agent_caller_access(auth)
capped: Final = grouped if isinstance(caller, UnrestrictedAgentAccess) else grouped & caller.agent_ids
context: Final = auth.managed_agent_context
if context is None or context.mode == "autonomous":
return RestrictedAgentAccess(capped)

View file

@ -237,6 +237,29 @@ class TestAgentRequestHandler:
frozenset()
)
async def test_managed_agent_acting_for_a_user_is_capped_at_the_invoking_teams_agents(self):
"""The managed path must honour the invoking team's ceiling the same way the unmanaged path does:
the agent's own policy grants alpha and beta, but the human who invoked it reaches only beta."""
from litellm.types.agents import AgentResponse
managed: Final = UserAPIKeyAuth(api_key="test-key", user_id="test-user", agent_id="actor")
managed.managed_agent_policy = AgentResponse(
agent_id="actor",
agent_name="Actor",
agent_card_params={},
object_permission={"object_permission_id": "own", "agents": ["agent-alpha", "agent-beta"]},
)
managed.agent_caller = AgentCaller(user_id="alice", team_id="callers")
with patch.object( # test-quality-ok: the team resolver reads proxy_server globals with no injection seam
AgentRequestHandler,
"_get_allowed_agents_for_team",
self._team_grants({"callers": RestrictedAgentAccess(frozenset({"agent-beta"}))}),
):
assert await AgentRequestHandler.resolve_agent_access(managed) == RestrictedAgentAccess(
frozenset({"agent-beta"})
)
async def test_agent_key_acting_for_an_ungranted_caller_keeps_its_own_agents(self):
agent_key: Final = self._key_granting(["agent-alpha"], agent_id="caller-agent")
agent_key.agent_caller = AgentCaller(user_id="alice", team_id="callers")