fix(mcp): cap a managed agent's servers and tools at the invoking caller

managed_agent_servers and managed_agent_tools returned the agent's own
grants without the agent_caller ceiling the unmanaged resolvers apply, so
a managed agent reached MCP servers and tools the echoed caller could not.
Call the existing ceiling helpers on both axes.
This commit is contained in:
Joshua Valluru 2026-09-29 19:17:53 -07:00
parent e9f6f5cd9c
commit ba5dd6365d
2 changed files with 50 additions and 2 deletions

View file

@ -29,7 +29,9 @@ async def managed_agent_servers(auth: UserAPIKeyAuth) -> tuple[str, ...]:
frozenset(global_mcp_server_manager.expand_permission_list(sorted(ceiling.mcp_server_ids)))
for ceiling in ceilings
)
own: Final = frozenset(server for server in base if all(server in ceiling for ceiling in expanded))
grouped: Final = frozenset(server for server in base if all(server in ceiling for ceiling in expanded))
caller_capped, _ = await MCPRequestHandler._apply_agent_caller_ceiling(sorted(grouped), auth)
own: Final = frozenset(caller_capped)
context: Final = auth.managed_agent_context
if context is None or context.mode == "autonomous":
return tuple(sorted(own))
@ -52,7 +54,9 @@ async def managed_agent_tools(server_id: str, auth: UserAPIKeyAuth) -> list[str]
if server_id not in await managed_agent_servers(auth):
return []
try:
own: Final = await MCPRequestHandler.get_agent_tool_permissions_for_server(server_id, auth)
granted: Final = await MCPRequestHandler.get_agent_tool_permissions_for_server(server_id, auth)
capped: Final = await MCPRequestHandler._apply_agent_caller_tool_ceiling(granted, server_id, auth)
own: Final = list(capped) if capped is not None else None
context: Final = auth.managed_agent_context
if context is None or context.mode == "autonomous":
return own

View file

@ -446,3 +446,47 @@ async def test_managed_mcp_rejects_unavailable_authoritative_entitlements(
assert failure.value.status_code == 503
client.db.litellm_mcpservertable.find_many.assert_not_called()
client.db.litellm_mcptoolsettable.find_many.assert_not_called()
@pytest.mark.asyncio
async def test_managed_agent_mcp_access_is_capped_at_the_invoking_callers_grants(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""The managed MCP path must honour the agent_caller ceiling the same way the unmanaged path does:
the agent's own policy grants slack and linear, but the team echoed back on the request reaches
only slack, so the agent may use slack alone."""
from litellm.proxy._types import AgentCaller
monkeypatch.setattr(
MCPRequestHandler,
"_get_allowed_mcp_servers_for_team",
AsyncMock(return_value=["slack"]),
)
monkeypatch.setattr(
MCPRequestHandler,
"_apply_user_server_ceiling",
AsyncMock(side_effect=lambda servers, _auth: (tuple(servers), False)),
)
monkeypatch.setattr(
MCPRequestHandler,
"_get_team_object_permission",
AsyncMock(
return_value=LiteLLM_ObjectPermissionTable(
object_permission_id="caller-team-permissions",
mcp_servers=["slack"],
mcp_tool_permissions={"slack": ["read"]},
)
),
)
monkeypatch.setattr(
MCPRequestHandler,
"_apply_user_tool_ceiling",
AsyncMock(side_effect=lambda tools, _server_id, _auth: tools),
)
auth: Final = actor(("read", "write"))
auth.agent_caller = AgentCaller(user_id="alice", team_id="callers")
assert set(await MCPRequestHandler.get_allowed_mcp_servers(auth)) == {"slack"}
assert await MCPRequestHandler.get_allowed_tools_for_server("slack", auth) == ["read"]