feat(mcp): enforce delegated-user ceiling for agent-OBO tool calls

A delegated (agent-on-behalf-of-user) MCP tool call was ceiling-enforced
only against the agent key, so a delegated run could reach any tool the
agent could, regardless of what the triggering user is allowed.

check_tool_permission_for_delegated_user runs in pre_call_tool_check right
after the existing key/team check. Because the agent's ceiling is already
enforced there, additionally requiring the delegated user to permit the
server and tool makes the effective permission the intersection of the two
principals. A non-delegated call early-returns and is byte-identical.

The delegated user's reach is resolved across ALL of their auth contexts
(their own object-permission plus one per team they belong to) and unioned,
so a user whose MCP access is granted directly, through toolsets/access
groups, or through team/group membership is counted rather than
under-counted. resolve_delegated_user_contexts builds the contexts (reusing
the fail-closed reload_admitted_user for the own context); the same
get_allowed_mcp_servers / get_allowed_tools_for_server primitives the key
path uses run per context, so no permission logic is duplicated. Only a
context that actually reaches a server contributes its tool grant, so a
team that cannot reach a server does not leak allow-all for it. Fail-closed
throughout; reload_admitted_user is promoted to public for the reuse.

Part of LIT-4448 (build item 4).

litellm/proxy/_experimental/mcp_server/auth/user_api_key_auth_mcp.py

@@ -703,12 +703,12 @@ class MCPRequestHandler:
             case "key_hash":
                 return await MCPRequestHandler._reload_admitted_key(identity.subject)
             case "user_id":
-                return await MCPRequestHandler._reload_admitted_user(identity.subject)
+                return await MCPRequestHandler.reload_admitted_user(identity.subject)
             case _:
                 assert_never(identity.subject_type)
 
     @staticmethod
-    async def _reload_admitted_user(user_id: str) -> UserAPIKeyAuth:
+    async def reload_admitted_user(user_id: str) -> UserAPIKeyAuth:
         """Reload the live user an interactively-minted envelope references and admit them as
         themselves.
 
@@ -771,6 +771,32 @@ class MCPRequestHandler:
             object_permission_id=user_object.object_permission_id,
         )
 
+    @staticmethod
+    async def resolve_delegated_user_contexts(user_id: str) -> list[UserAPIKeyAuth]:
+        """The auth contexts spanning a delegated user's TOTAL MCP reach: their own object-permission
+        context, plus one per team they belong to (a key on that team inherits the team's grants).
+
+        ``get_allowed_mcp_servers`` / ``get_allowed_tools_for_server`` computed over the UNION of these
+        contexts give the user's full cross-team reach, which a single ``UserAPIKeyAuth`` (one team)
+        cannot express; this is what lets a delegated user whose MCP access is granted through team or
+        group membership be counted rather than under-counted. The own context reuses the fail-closed
+        ``reload_admitted_user`` resolver; a listed team with no MCP grants simply contributes nothing
+        to the union, and an unresolvable user still fails closed there.
+        """
+        from litellm.proxy.auth.auth_checks import get_user_object
+        from litellm.proxy.proxy_server import prisma_client, user_api_key_cache
+
+        own_context = await MCPRequestHandler.reload_admitted_user(user_id)
+        user_object = await get_user_object(
+            user_id=user_id,
+            prisma_client=prisma_client,
+            user_api_key_cache=user_api_key_cache,
+            user_id_upsert=False,
+        )
+        team_ids = list(user_object.teams) if user_object is not None and user_object.teams else []
+        team_contexts = [UserAPIKeyAuth(user_id=user_id, team_id=team_id) for team_id in team_ids]
+        return [own_context, *team_contexts]
+
     @staticmethod
     async def _reload_admitted_key(key_hash: str) -> UserAPIKeyAuth:
         """Reload the live key record an admitted envelope references and re-check live policy.

litellm/proxy/_experimental/mcp_server/mcp_server_manager.py

@@ -204,6 +204,33 @@ def _blank_to_none(value: str | None) -> str | None:
     return value.strip() or None
 
 
+def _delegated_reach_denial(
+    server_id: str,
+    tool_name: str,
+    per_context_servers: list[list[str]],
+    per_context_tools: list[list[str] | None],
+) -> Literal["server", "tool"] | None:
+    """Why a delegated call is denied by the user's ceiling, or ``None`` if the user permits it.
+
+    ``per_context_servers`` / ``per_context_tools`` are the allowed servers and per-server allowed
+    tools resolved for each of the delegated user's auth contexts (own + one per team). Reach is the
+    UNION across contexts (the user's total cross-team access): the server must be reachable in some
+    context. Only a context that actually reaches the server contributes its tool grant, since a
+    context that cannot reach the server returns "no restriction" (``None``) for it and would
+    otherwise be misread as allow-all; among the reaching contexts, ``None`` means allow-all and
+    otherwise the tool must appear in the union of their restricted lists.
+    """
+    reaching_context_tools = [
+        tools for servers, tools in zip(per_context_servers, per_context_tools) if server_id in servers
+    ]
+    if not reaching_context_tools:
+        return "server"
+    if any(tools is None for tools in reaching_context_tools):
+        return None
+    allowed_tools = frozenset(tool for tools in reaching_context_tools if tools for tool in tools)
+    return None if tool_name in allowed_tools else "tool"
+
+
 def _uses_issuer_anchor(manual_issuer: str | None, is_discovery_auth_type: bool) -> bool:
     """Whether the endpoints are authoritatively anchored to an admin-pinned issuer (RFC 8414 §3.3).
 
@@ -4001,6 +4028,55 @@ class MCPServerManager:
                 },
             )
 
+    async def check_tool_permission_for_delegated_user(
+        self,
+        tool_name: str,
+        server: MCPServer,
+        user_api_key_auth: UserAPIKeyAuth | None,
+    ) -> None:
+        """For an agent-delegated (on-behalf-of) call, enforce the DELEGATED USER's ceiling as well.
+
+        The preceding key/team check already enforced the agent key's ceiling, so requiring the
+        delegated user to ALSO permit the server and tool makes the effective permission the
+        intersection of the two principals: a delegated run can reach only what both the agent and the
+        triggering user may reach. A non-delegated call early-returns, so its behavior is unchanged.
+
+        The delegated user's reach is resolved across ALL of their auth contexts (own object-permission
+        plus one per team they belong to; ``resolve_delegated_user_contexts``) and UNIONED, so a user
+        whose MCP access is granted directly, through toolsets/access groups, or through team/group
+        membership is counted rather than under-counted. The same ``get_allowed_mcp_servers`` /
+        ``get_allowed_tools_for_server`` primitives the key path uses run per context, so no permission
+        logic is duplicated. Fail-closed: an unresolvable user raises; server-reachability is checked
+        explicitly because the tool predicate allow-alls a server the user has no restriction on.
+        """
+        if user_api_key_auth is None or user_api_key_auth.delegated_user_id is None:
+            return
+        from litellm.proxy._experimental.mcp_server.auth.user_api_key_auth_mcp import (
+            MCPRequestHandler,
+        )
+
+        contexts = await MCPRequestHandler.resolve_delegated_user_contexts(user_api_key_auth.delegated_user_id)
+        per_context_servers = [await MCPRequestHandler.get_allowed_mcp_servers(context) for context in contexts]
+        per_context_tools = [
+            await MCPRequestHandler.get_allowed_tools_for_server(server_id=server.server_id, user_api_key_auth=context)
+            for context in contexts
+        ]
+        denial = _delegated_reach_denial(server.server_id, tool_name, per_context_servers, per_context_tools)
+        if denial == "server":
+            raise HTTPException(
+                status_code=403,
+                detail={
+                    "error": f"Delegated user is not allowed to reach server '{server.name}'. A delegated run is limited to the intersection of the agent's and the user's access."
+                },
+            )
+        if denial == "tool":
+            raise HTTPException(
+                status_code=403,
+                detail={
+                    "error": f"Delegated user is not allowed to call tool '{tool_name}' on server '{server.name}'. A delegated run is limited to the intersection of the agent's and the user's access."
+                },
+            )
+
     async def _call_openapi_tool_handler(
         self,
         server: MCPServer,
@@ -4085,13 +4161,21 @@ class MCPServerManager:
                 },
             )
 
-        ## check tool-level permissions from object_permission
+        ## check tool-level permissions from object_permission (the agent key/team ceiling)
         await self.check_tool_permission_for_key_team(
             tool_name=name,
             server=server,
             user_api_key_auth=user_api_key_auth,
         )
 
+        ## for a delegated (on-behalf-of) call, also enforce the delegated user's ceiling, so the
+        ## effective permission is the intersection of the agent and the user
+        await self.check_tool_permission_for_delegated_user(
+            tool_name=name,
+            server=server,
+            user_api_key_auth=user_api_key_auth,
+        )
+
         ## filter parameters based on allowed_params configuration
         self.validate_allowed_params(
             tool_name=name,

tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py

@@ -4659,6 +4659,129 @@ class TestMCPServerManager:
             user_api_key_auth=user_auth,
         )
 
+    # ── delegated-user (agent-OBO) intersection enforcement ──────────────────────
+
+    @pytest.mark.asyncio
+    async def test_delegated_check_is_a_noop_for_a_non_delegated_call(self, monkeypatch):
+        """A call with no delegated_user_id must not resolve any user or change behavior."""
+        from litellm.proxy._experimental.mcp_server.auth.user_api_key_auth_mcp import MCPRequestHandler
+        from litellm.proxy._types import UserAPIKeyAuth
+
+        async def _must_not_run(user_id):
+            raise AssertionError("non-delegated calls must not resolve a delegated user")
+
+        monkeypatch.setattr(MCPRequestHandler, "resolve_delegated_user_contexts", _must_not_run)
+        manager = MCPServerManager()
+        server = MCPServer(server_id="srv-1", name="S", transport=MCPTransport.http)
+        # No raise, and _must_not_run is never called.
+        await manager.check_tool_permission_for_delegated_user(
+            tool_name="anything",
+            server=server,
+            user_api_key_auth=UserAPIKeyAuth(api_key="sk-agent", user_id="agent", delegated_user_id=None),
+        )
+
+    def _stub_delegated_user(self, monkeypatch, *, mcp_servers, mcp_tool_permissions):
+        from litellm.proxy._experimental.mcp_server.auth.user_api_key_auth_mcp import MCPRequestHandler
+        from litellm.proxy._types import LiteLLM_ObjectPermissionTable, UserAPIKeyAuth
+
+        own_context = UserAPIKeyAuth(
+            user_id="alice",
+            object_permission=LiteLLM_ObjectPermissionTable(
+                object_permission_id="perm-alice",
+                mcp_servers=mcp_servers,
+                mcp_tool_permissions=mcp_tool_permissions,
+            ),
+        )
+
+        async def _contexts(user_id):
+            assert user_id == "alice"
+            return [own_context]
+
+        monkeypatch.setattr(MCPRequestHandler, "resolve_delegated_user_contexts", _contexts)
+
+    @pytest.mark.asyncio
+    async def test_delegated_user_within_ceiling_passes(self, monkeypatch):
+        from litellm.proxy._types import UserAPIKeyAuth
+
+        self._stub_delegated_user(monkeypatch, mcp_servers=["srv-1"], mcp_tool_permissions={"srv-1": ["read"]})
+        manager = MCPServerManager()
+        server = MCPServer(server_id="srv-1", name="S", transport=MCPTransport.http)
+        agent = UserAPIKeyAuth(api_key="sk-agent", user_id="agent", delegated_user_id="alice")
+        # The delegated user may reach srv-1 and call read -> no raise.
+        await manager.check_tool_permission_for_delegated_user(tool_name="read", server=server, user_api_key_auth=agent)
+
+    @pytest.mark.asyncio
+    async def test_delegated_user_denied_tool_raises_403(self, monkeypatch):
+        from litellm.proxy._types import UserAPIKeyAuth
+
+        self._stub_delegated_user(monkeypatch, mcp_servers=["srv-1"], mcp_tool_permissions={"srv-1": ["read"]})
+        manager = MCPServerManager()
+        server = MCPServer(server_id="srv-1", name="S", transport=MCPTransport.http)
+        agent = UserAPIKeyAuth(api_key="sk-agent", user_id="agent", delegated_user_id="alice")
+        # The agent may permit write, but the delegated user does not -> intersection denies it.
+        with pytest.raises(HTTPException) as exc:
+            await manager.check_tool_permission_for_delegated_user(tool_name="write", server=server, user_api_key_auth=agent)
+        assert exc.value.status_code == 403
+        assert "write" in exc.value.detail["error"]
+
+    @pytest.mark.asyncio
+    async def test_delegated_user_cannot_reach_server_raises_403(self, monkeypatch):
+        from litellm.proxy._types import UserAPIKeyAuth
+
+        self._stub_delegated_user(monkeypatch, mcp_servers=["other-srv"], mcp_tool_permissions={"other-srv": ["read"]})
+        manager = MCPServerManager()
+        server = MCPServer(server_id="srv-1", name="S", transport=MCPTransport.http)
+        agent = UserAPIKeyAuth(api_key="sk-agent", user_id="agent", delegated_user_id="alice")
+        # The agent may reach srv-1, but the delegated user cannot -> intersection denies the server.
+        with pytest.raises(HTTPException) as exc:
+            await manager.check_tool_permission_for_delegated_user(tool_name="read", server=server, user_api_key_auth=agent)
+        assert exc.value.status_code == 403
+        assert "srv-1" in exc.value.detail["error"] or "S" in exc.value.detail["error"]
+
+    def test_delegated_reach_denial_unions_across_contexts(self):
+        """The pure union math: a delegated user's reach is the union across their contexts (own +
+        each team), so a grant from ANY team counts."""
+        from litellm.proxy._experimental.mcp_server.mcp_server_manager import _delegated_reach_denial
+
+        # team-A reaches srv-1/read; team-B reaches srv-2/write. The union reaches both.
+        servers = [["srv-1"], ["srv-2"]]
+        tools = [["read"], ["write"]]
+        assert _delegated_reach_denial("srv-1", "read", servers, tools) is None  # via team-A
+        assert _delegated_reach_denial("srv-2", "write", servers, tools) is None  # via team-B (union)
+        assert _delegated_reach_denial("srv-3", "read", servers, tools) == "server"  # no context reaches srv-3
+        # server reachable (union) but the tool is in no context's list.
+        assert _delegated_reach_denial("srv-1", "delete", [["srv-1"], ["srv-1"]], [["read"], ["write"]]) == "tool"
+        # a context with no tool restriction (None) means allow-all for that server.
+        assert _delegated_reach_denial("srv-1", "anything", [["srv-1"]], [None]) is None
+        # a context that CANNOT reach the server must not contribute its (unrestricted) tools as
+        # allow-all: here only the srv-1 context (tools ["read"]) counts, so write is denied even
+        # though the srv-2 context returned None for srv-1.
+        assert _delegated_reach_denial("srv-1", "write", [["srv-1"], ["srv-2"]], [["read"], None]) == "tool"
+
+    @pytest.mark.asyncio
+    async def test_resolve_delegated_user_contexts_includes_each_team(self, monkeypatch):
+        """resolve_delegated_user_contexts returns the own context plus one per team the user is in, so
+        the union covers team/group-granted access."""
+        from types import SimpleNamespace
+
+        import litellm.proxy.auth.auth_checks as auth_checks
+        from litellm.proxy._experimental.mcp_server.auth.user_api_key_auth_mcp import MCPRequestHandler
+        from litellm.proxy._types import UserAPIKeyAuth
+
+        async def _own(user_id):
+            return UserAPIKeyAuth(user_id=user_id, team_id=None)
+
+        async def _get_user_object(**kwargs):
+            return SimpleNamespace(user_id="alice", teams=["team-a", "team-b"])
+
+        monkeypatch.setattr(MCPRequestHandler, "reload_admitted_user", _own)
+        monkeypatch.setattr(auth_checks, "get_user_object", _get_user_object)
+
+        contexts = await MCPRequestHandler.resolve_delegated_user_contexts("alice")
+        assert len(contexts) == 3
+        assert contexts[0].team_id is None  # own context
+        assert sorted(c.team_id for c in contexts[1:]) == ["team-a", "team-b"]
+
     @pytest.mark.asyncio
     async def test_allowed_tools_with_mixed_prefixed_and_unprefixed_names(self):
         """