add_user_to_group
Assign a user to a group in Keycloak identity management by specifying user ID, group ID, and optional realm to manage access permissions.
Instructions
Add a user to a group.
Args:
user_id: User ID
group_id: Group ID
realm: Target realm (uses default if not specified)
Returns:
Status messageInput Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| user_id | Yes | ||
| group_id | Yes | ||
| realm | No |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- src/tools/group_tools.py:170-188 (handler)The core handler function for the 'add_user_to_group' tool. It is decorated with @mcp.tool() which serves as the registration. The function makes a PUT request to the Keycloak API endpoint /users/{user_id}/groups/{group_id} using the KeycloakClient to add the user to the group and returns a success message.
@mcp.tool() async def add_user_to_group( user_id: str, group_id: str, realm: Optional[str] = None ) -> Dict[str, str]: """ Add a user to a group. Args: user_id: User ID group_id: Group ID realm: Target realm (uses default if not specified) Returns: Status message """ await client._make_request( "PUT", f"/users/{user_id}/groups/{group_id}", realm=realm ) return {"status": "added", "message": f"User {user_id} added to group {group_id}"} - src/tools/keycloak_client.py:59-108 (helper)Supporting helper method _make_request in KeycloakClient class, used by the tool handler to perform authenticated HTTP requests to the Keycloak Admin API, handling token refresh and realm targeting.
async def _make_request( self, method: str, endpoint: str, data: Optional[Dict] = None, params: Optional[Dict] = None, skip_realm: bool = False, realm: Optional[str] = None, ) -> Any: """Make authenticated request to Keycloak API""" if skip_realm: url = f"{self.server_url}/auth/admin{endpoint}" else: # Use provided realm or fall back to configured realm target_realm = realm if realm is not None else self.realm_name url = f"{self.server_url}/auth/admin/realms/{target_realm}{endpoint}" try: client = await self._ensure_client() headers = await self._get_headers() response = await client.request( method=method, url=url, headers=headers, json=data, params=params, ) # If token expired, refresh and retry if response.status_code == 401: await self._get_token() headers = await self._get_headers() response = await client.request( method=method, url=url, headers=headers, json=data, params=params, ) response.raise_for_status() if response.content: return response.json() return None except httpx.RequestError as e: raise Exception(f"Keycloak API request failed: {str(e)}")