admin/mcp-nextcloud
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: Update user/groups API to OCS v2

Browse Source
This commit is contained in:
Chris Coutinho
2025-10-15 00:05:22 +02:00
parent 898c2e72ae
commit 7a4a31b52d
3 changed files with 35 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nextcloud_mcp_server/client/users.py
+24 -46
View File
@@ -10,7 +10,7 @@ class UsersClient(BaseNextcloudClient):
self, additional_headers: Optional[Dict[str, str]] = None self, additional_headers: Optional[Dict[str, str]] = None
) -> Dict[str, str]: ) -> Dict[str, str]:
"""Get standard headers required for User API calls.""" """Get standard headers required for User API calls."""
headers = {"OCS-APIRequest": "true"} headers = {"OCS-APIRequest": "true", "Accept": "application/json"}
if additional_headers: if additional_headers:
headers.update(additional_headers) headers.update(additional_headers)
return headers return headers
@@ -49,7 +49,7 @@ class UsersClient(BaseNextcloudClient):
headers = self._get_user_headers() headers = self._get_user_headers()
await self._make_request( await self._make_request(
"POST", "/ocs/v1.php/cloud/users", data=data, headers=headers "POST", "/ocs/v2.php/cloud/users", data=data, headers=headers
) )
async def search_users( async def search_users(
@@ -71,18 +71,11 @@ class UsersClient(BaseNextcloudClient):
headers = self._get_user_headers() headers = self._get_user_headers()
response = await self._make_request( response = await self._make_request(
"GET", "/ocs/v1.php/cloud/users", params=params, headers=headers "GET", "/ocs/v2.php/cloud/users", params=params, headers=headers
) )
# The API returns XML, which is parsed into a dict. # The v2 API returns JSON with users as a direct list under data.users
# The user IDs are under ocs.data.users.element (can be a list or a single string)
data = response.json()["ocs"]["data"] data = response.json()["ocs"]["data"]
if "users" in data and "element" in data["users"]: return data.get("users", [])
elements = data["users"]["element"]
if isinstance(elements, list):
return elements
elif isinstance(elements, str):
return [elements]
return []
async def get_user_details(self, userid: str) -> UserDetails: async def get_user_details(self, userid: str) -> UserDetails:
""" """
@@ -90,7 +83,7 @@ class UsersClient(BaseNextcloudClient):
""" """
headers = self._get_user_headers() headers = self._get_user_headers()
response = await self._make_request( response = await self._make_request(
"GET", f"/ocs/v1.php/cloud/users/{userid}", headers=headers "GET", f"/ocs/v2.php/cloud/users/{userid}", headers=headers
) )
return UserDetails(**response.json()["ocs"]["data"]) return UserDetails(**response.json()["ocs"]["data"])
@@ -101,7 +94,7 @@ class UsersClient(BaseNextcloudClient):
data = {"key": key, "value": value} data = {"key": key, "value": value}
headers = self._get_user_headers() headers = self._get_user_headers()
await self._make_request( await self._make_request(
"PUT", f"/ocs/v1.php/cloud/users/{userid}", data=data, headers=headers "PUT", f"/ocs/v2.php/cloud/users/{userid}", data=data, headers=headers
) )
async def get_editable_user_fields(self) -> List[str]: async def get_editable_user_fields(self) -> List[str]:
@@ -110,16 +103,11 @@ class UsersClient(BaseNextcloudClient):
""" """
headers = self._get_user_headers() headers = self._get_user_headers()
response = await self._make_request( response = await self._make_request(
"GET", "/ocs/v1.php/cloud/user/fields", headers=headers "GET", "/ocs/v2.php/cloud/user/fields", headers=headers
) )
# The v2 API returns data as a direct list
data = response.json()["ocs"]["data"] data = response.json()["ocs"]["data"]
if "element" in data: return data if isinstance(data, list) else []
elements = data["element"]
if isinstance(elements, list):
return elements
elif isinstance(elements, str):
return [elements]
return []
async def disable_user(self, userid: str) -> None: async def disable_user(self, userid: str) -> None:
""" """
@@ -127,7 +115,7 @@ class UsersClient(BaseNextcloudClient):
""" """
headers = self._get_user_headers() headers = self._get_user_headers()
await self._make_request( await self._make_request(
"PUT", f"/ocs/v1.php/cloud/users/{userid}/disable", headers=headers "PUT", f"/ocs/v2.php/cloud/users/{userid}/disable", headers=headers
) )
async def enable_user(self, userid: str) -> None: async def enable_user(self, userid: str) -> None:
@@ -136,7 +124,7 @@ class UsersClient(BaseNextcloudClient):
""" """
headers = self._get_user_headers() headers = self._get_user_headers()
await self._make_request( await self._make_request(
"PUT", f"/ocs/v1.php/cloud/users/{userid}/enable", headers=headers "PUT", f"/ocs/v2.php/cloud/users/{userid}/enable", headers=headers
) )
async def delete_user(self, userid: str) -> None: async def delete_user(self, userid: str) -> None:
@@ -145,7 +133,7 @@ class UsersClient(BaseNextcloudClient):
""" """
headers = self._get_user_headers() headers = self._get_user_headers()
await self._make_request( await self._make_request(
"DELETE", f"/ocs/v1.php/cloud/users/{userid}", headers=headers "DELETE", f"/ocs/v2.php/cloud/users/{userid}", headers=headers
) )
async def get_user_groups(self, userid: str) -> List[str]: async def get_user_groups(self, userid: str) -> List[str]:
@@ -154,16 +142,11 @@ class UsersClient(BaseNextcloudClient):
""" """
headers = self._get_user_headers() headers = self._get_user_headers()
response = await self._make_request( response = await self._make_request(
"GET", f"/ocs/v1.php/cloud/users/{userid}/groups", headers=headers "GET", f"/ocs/v2.php/cloud/users/{userid}/groups", headers=headers
) )
# The v2 API returns groups as a direct list under data.groups
data = response.json()["ocs"]["data"] data = response.json()["ocs"]["data"]
if "groups" in data and "element" in data["groups"]: return data.get("groups", [])
elements = data["groups"]["element"]
if isinstance(elements, list):
return elements
elif isinstance(elements, str):
return [elements]
return []
async def add_user_to_group(self, userid: str, groupid: str) -> None: async def add_user_to_group(self, userid: str, groupid: str) -> None:
""" """
@@ -173,7 +156,7 @@ class UsersClient(BaseNextcloudClient):
headers = self._get_user_headers() headers = self._get_user_headers()
await self._make_request( await self._make_request(
"POST", "POST",
f"/ocs/v1.php/cloud/users/{userid}/groups", f"/ocs/v2.php/cloud/users/{userid}/groups",
data=data, data=data,
headers=headers, headers=headers,
) )
@@ -186,7 +169,7 @@ class UsersClient(BaseNextcloudClient):
headers = self._get_user_headers() headers = self._get_user_headers()
await self._make_request( await self._make_request(
"DELETE", "DELETE",
f"/ocs/v1.php/cloud/users/{userid}/groups", f"/ocs/v2.php/cloud/users/{userid}/groups",
data=data, data=data,
headers=headers, headers=headers,
) )
@@ -199,7 +182,7 @@ class UsersClient(BaseNextcloudClient):
headers = self._get_user_headers() headers = self._get_user_headers()
await self._make_request( await self._make_request(
"POST", "POST",
f"/ocs/v1.php/cloud/users/{userid}/subadmins", f"/ocs/v2.php/cloud/users/{userid}/subadmins",
data=data, data=data,
headers=headers, headers=headers,
) )
@@ -212,7 +195,7 @@ class UsersClient(BaseNextcloudClient):
headers = self._get_user_headers() headers = self._get_user_headers()
await self._make_request( await self._make_request(
"DELETE", "DELETE",
f"/ocs/v1.php/cloud/users/{userid}/subadmins", f"/ocs/v2.php/cloud/users/{userid}/subadmins",
data=data, data=data,
headers=headers, headers=headers,
) )
@@ -223,16 +206,11 @@ class UsersClient(BaseNextcloudClient):
""" """
headers = self._get_user_headers() headers = self._get_user_headers()
response = await self._make_request( response = await self._make_request(
"GET", f"/ocs/v1.php/cloud/users/{userid}/subadmins", headers=headers "GET", f"/ocs/v2.php/cloud/users/{userid}/subadmins", headers=headers
) )
# The v2 API returns data as a direct list
data = response.json()["ocs"]["data"] data = response.json()["ocs"]["data"]
if "element" in data: return data if isinstance(data, list) else []
elements = data["element"]
if isinstance(elements, list):
return elements
elif isinstance(elements, str):
return [elements]
return []
async def resend_welcome_email(self, userid: str) -> None: async def resend_welcome_email(self, userid: str) -> None:
""" """
@@ -240,5 +218,5 @@ class UsersClient(BaseNextcloudClient):
""" """
headers = self._get_user_headers() headers = self._get_user_headers()
await self._make_request( await self._make_request(
"POST", f"/ocs/v1.php/cloud/users/{userid}/welcome", headers=headers "POST", f"/ocs/v2.php/cloud/users/{userid}/welcome", headers=headers
) )
nextcloud_mcp_server/models/users.py
+6 -4
View File
@@ -1,5 +1,5 @@
from typing import List, Optional from typing import Any, Dict, List, Optional, Union
from pydantic import BaseModel, Field from pydantic import BaseModel, ConfigDict, Field
class User(BaseModel): class User(BaseModel):
@@ -18,10 +18,12 @@ class User(BaseModel):
class UserDetails(BaseModel): class UserDetails(BaseModel):
"""Model for retrieving detailed user information.""" """Model for retrieving detailed user information."""
model_config = ConfigDict(populate_by_name=True)
enabled: bool enabled: bool
id: str id: str
quota: str quota: Union[str, Dict[str, Any]] # Can be string or quota object
email: str email: Optional[str] = None # Can be null
displayname: str = Field( displayname: str = Field(
alias="display-name" alias="display-name"
) # Handle both displayname and display-name ) # Handle both displayname and display-name
tests/server/test_users_api.py
+5 -5
View File
@@ -5,7 +5,7 @@ from nextcloud_mcp_server.client import NextcloudClient
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_create_and_delete_user(nc_client: NextcloudClient): async def test_create_and_delete_user(nc_client: NextcloudClient):
userid = "testuser1" userid = "testuser1"
password = "testpassword1" password = "SecureTestPassword123!"
display_name = "Test User One" display_name = "Test User One"
email = "test1@example.com" email = "test1@example.com"
@@ -37,7 +37,7 @@ async def test_create_and_delete_user(nc_client: NextcloudClient):
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_update_user_field(nc_client: NextcloudClient): async def test_update_user_field(nc_client: NextcloudClient):
userid = "testuser2" userid = "testuser2"
password = "testpassword2" password = "SecureTestPassword123!"
display_name = "Test User Two" display_name = "Test User Two"
email = "test2@example.com" email = "test2@example.com"
@@ -60,7 +60,7 @@ async def test_update_user_field(nc_client: NextcloudClient):
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_user_groups(nc_client: NextcloudClient): async def test_user_groups(nc_client: NextcloudClient):
userid = "testuser3" userid = "testuser3"
password = "testpassword3" password = "SecureTestPassword123!"
groupid = "testgroup" groupid = "testgroup"
await nc_client.users.create_user(userid=userid, password=password) await nc_client.users.create_user(userid=userid, password=password)
@@ -81,7 +81,7 @@ async def test_user_groups(nc_client: NextcloudClient):
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_user_subadmins(nc_client: NextcloudClient): async def test_user_subadmins(nc_client: NextcloudClient):
userid = "testuser4" userid = "testuser4"
password = "testpassword4" password = "SecureTestPassword123!"
groupid = "subadmingroup" groupid = "subadmingroup"
await nc_client.users.create_user(userid=userid, password=password) await nc_client.users.create_user(userid=userid, password=password)
@@ -102,7 +102,7 @@ async def test_user_subadmins(nc_client: NextcloudClient):
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_disable_enable_user(nc_client: NextcloudClient): async def test_disable_enable_user(nc_client: NextcloudClient):
userid = "testuser5" userid = "testuser5"
password = "testpassword5" password = "SecureTestPassword123!"
await nc_client.users.create_user(userid=userid, password=password) await nc_client.users.create_user(userid=userid, password=password)