feat: update MCP CRUD endpoints to support mcp_server_manager role

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

litellm/proxy/management_endpoints/mcp_management_endpoints.py

@@ -1273,16 +1273,26 @@ if MCP_AVAILABLE:
         # Validate and normalize payload fields
         validate_and_normalize_mcp_server_payload(payload)
 
-        # AuthZ - restrict only proxy admins to create mcp servers
-        if LitellmUserRoles.PROXY_ADMIN != user_api_key_dict.user_role:
-            raise HTTPException(
-                status_code=status.HTTP_403_FORBIDDEN,
-                detail={
-                    "error": "User does not have permission to create mcp servers. You can only create mcp servers if you are a PROXY_ADMIN."
-                },
+        # AuthZ - proxy admins or team MCP managers can create MCP servers
+        is_proxy_admin = LitellmUserRoles.PROXY_ADMIN == user_api_key_dict.user_role
+        manager_team_id: Optional[str] = None
+
+        if not is_proxy_admin:
+            # Check if the user is an MCP manager for the specified team
+            if payload.team_id is None:
+                raise HTTPException(
+                    status_code=status.HTTP_400_BAD_REQUEST,
+                    detail={
+                        "error": "team_id is required when creating MCP servers as a team MCP manager."
+                    },
+                )
+            manager_team_id = await _assert_can_manage_team_mcp_server(
+                user_api_key_dict=user_api_key_dict,
+                team_id=payload.team_id,
             )
-        elif payload.server_id is not None:
-            # fail if the mcp server with id already exists
+
+        # Fail if the MCP server with this id already exists
+        if payload.server_id is not None:
             mcp_server = await get_mcp_server(prisma_client, payload.server_id)
             if mcp_server is not None:
                 raise HTTPException(
@@ -1291,7 +1301,8 @@ if MCP_AVAILABLE:
                         "error": f"MCP Server with id {payload.server_id} already exists. Cannot create another."
                     },
                 )
-        elif (
+
+        if (
             SpecialMCPServerName.all_team_servers == payload.server_id
             or SpecialMCPServerName.all_proxy_servers == payload.server_id
         ):
@@ -1322,6 +1333,40 @@ if MCP_AVAILABLE:
 
             # Ensure registry is up to date by reloading from database
             await global_mcp_server_manager.reload_servers_from_database()
+
+            # If created by an MCP manager, auto-assign the server to their team
+            if manager_team_id is not None:
+                from litellm.proxy.proxy_server import user_api_key_cache
+
+                team_obj = await get_team_object(
+                    team_id=manager_team_id,
+                    prisma_client=prisma_client,
+                    user_api_key_cache=user_api_key_cache,
+                    check_db_only=False,
+                )
+                existing_permission_id = getattr(
+                    team_obj, "object_permission_id", None
+                )
+
+                # Read existing mcp_servers list and append the new server
+                existing_mcp_servers: list = []
+                if team_obj.object_permission is not None:
+                    existing_mcp_servers = (
+                        team_obj.object_permission.mcp_servers or []
+                    )
+                updated_mcp_servers = list(
+                    set(existing_mcp_servers + [new_mcp_server.server_id])
+                )
+
+                await handle_update_object_permission_common(
+                    data_json={
+                        "object_permission": {
+                            "mcp_servers": updated_mcp_servers,
+                        }
+                    },
+                    existing_object_permission_id=existing_permission_id,
+                    prisma_client=prisma_client,
+                )
         except Exception as e:
             verbose_proxy_logger.exception(f"Error creating mcp server: {str(e)}")
             raise HTTPException(
@@ -1540,15 +1585,11 @@ if MCP_AVAILABLE:
             "Database not connected. Connect a database to your proxy - https://docs.litellm.ai/docs/simple_proxy#managing-auth---virtual-keys"
         )
 
-        # Authz - restrict only admins to delete mcp servers
+        # Authz - proxy admins or team MCP managers can delete MCP servers
         if LitellmUserRoles.PROXY_ADMIN != user_api_key_dict.user_role:
-            raise HTTPException(
-                status_code=status.HTTP_403_FORBIDDEN,
-                detail={
-                    "error": "Call not allowed to delete MCP server. User is not a proxy admin. route={}".format(
-                        "DELETE /v1/mcp/server"
-                    )
-                },
+            await _assert_can_manage_team_mcp_server(
+                user_api_key_dict=user_api_key_dict,
+                server_id=server_id,
             )
 
         # try to delete the mcp server
@@ -1865,15 +1906,11 @@ if MCP_AVAILABLE:
         # Validate and normalize payload fields
         validate_and_normalize_mcp_server_payload(payload)
 
-        # Authz - restrict only admins to delete mcp servers
+        # Authz - proxy admins or team MCP managers can update MCP servers
         if LitellmUserRoles.PROXY_ADMIN != user_api_key_dict.user_role:
-            raise HTTPException(
-                status_code=status.HTTP_403_FORBIDDEN,
-                detail={
-                    "error": "Call not allowed to update MCP server. User is not a proxy admin. route={}".format(
-                        "PUT /v1/mcp/server"
-                    )
-                },
+            await _assert_can_manage_team_mcp_server(
+                user_api_key_dict=user_api_key_dict,
+                server_id=payload.server_id,
             )
 
         # try to update the mcp server

tests/test_litellm/proxy/management_endpoints/test_mcp_manager_role.py

@@ -185,3 +185,87 @@ class TestAssertCanManageTeamMcpServer:
                     user_api_key_dict=user_auth, server_id="server1"
                 )
             assert exc_info.value.status_code == 403
+
+
+@pytest.mark.asyncio
+class TestCreateMcpServerAsManager:
+    async def test_create_auto_assigns_to_team(self):
+        """MCP manager creating a server should auto-assign it to their team's ObjectPermissionTable."""
+        from litellm.proxy._types import NewMCPServerRequest
+        from litellm.proxy.management_endpoints.mcp_management_endpoints import (
+            add_mcp_server,
+        )
+
+        user_auth = UserAPIKeyAuth(
+            user_role=LitellmUserRoles.INTERNAL_USER,
+            user_id="user1",
+            api_key="sk-test",
+            team_id="team1",
+        )
+
+        payload = NewMCPServerRequest(
+            server_name="test-server",
+            url="https://example.com/mcp",
+            team_id="team1",
+        )
+
+        mock_team = LiteLLM_TeamTableCachedObj(
+            team_id="team1",
+            members_with_roles=[
+                Member(user_id="user1", role="mcp_server_manager"),
+            ],
+            object_permission_id="perm1",
+        )
+        mock_team.object_permission = MagicMock(mcp_servers=["existing_server"])
+
+        created_server = MagicMock()
+        created_server.server_id = "new_server_id"
+        created_server.credentials = None
+
+        mock_handle_update = AsyncMock()
+
+        with (
+            patch(
+                "litellm.proxy.management_endpoints.mcp_management_endpoints.get_prisma_client_or_throw",
+                return_value=MagicMock(),
+            ),
+            patch(
+                "litellm.proxy.management_endpoints.mcp_management_endpoints.validate_and_normalize_mcp_server_payload",
+            ),
+            patch(
+                "litellm.proxy.management_endpoints.mcp_management_endpoints._assert_can_manage_team_mcp_server",
+                AsyncMock(return_value="team1"),
+            ),
+            patch(
+                "litellm.proxy.management_endpoints.mcp_management_endpoints.get_mcp_server",
+                AsyncMock(return_value=None),
+            ),
+            patch(
+                "litellm.proxy.management_endpoints.mcp_management_endpoints.create_mcp_server",
+                AsyncMock(return_value=created_server),
+            ),
+            patch(
+                "litellm.proxy.management_endpoints.mcp_management_endpoints.global_mcp_server_manager",
+                MagicMock(
+                    add_server=AsyncMock(),
+                    reload_servers_from_database=AsyncMock(),
+                ),
+            ),
+            patch(
+                "litellm.proxy.management_endpoints.mcp_management_endpoints.get_team_object",
+                AsyncMock(return_value=mock_team),
+            ),
+            patch(
+                "litellm.proxy.management_endpoints.mcp_management_endpoints.handle_update_object_permission_common",
+                mock_handle_update,
+            ),
+        ):
+            await add_mcp_server(payload=payload, user_api_key_dict=user_auth)
+
+            # Verify handle_update_object_permission_common was called with merged server list
+            mock_handle_update.assert_called_once()
+            call_kwargs = mock_handle_update.call_args.kwargs
+            mcp_servers = call_kwargs["data_json"]["object_permission"]["mcp_servers"]
+            assert "existing_server" in mcp_servers
+            assert "new_server_id" in mcp_servers
+            assert call_kwargs["existing_object_permission_id"] == "perm1"