logout_user
Terminate all active sessions for a specified user in the mcp-keycloak server. Requires the user ID and optionally a target realm to complete the logout process effectively.
Instructions
Logout all sessions for a user.
Args:
user_id: The user's ID
realm: Target realm (uses default if not specified)
Returns:
Status message
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| realm | No | ||
| user_id | Yes |
Implementation Reference
- src/tools/user_tools.py:231-247 (handler)The handler function decorated with @mcp.tool(), implementing the logout_user tool by posting to Keycloak's user logout endpoint.@mcp.tool() async def logout_user(user_id: str, realm: Optional[str] = None) -> Dict[str, str]: """ Logout all sessions for a user. Args: user_id: The user's ID realm: Target realm (uses default if not specified) Returns: Status message """ await client._make_request("POST", f"/users/{user_id}/logout", realm=realm) return { "status": "success", "message": f"User {user_id} logged out from all sessions", }
- src/tools/keycloak_client.py:59-108 (helper)The _make_request helper method in KeycloakClient used by logout_user to perform the authenticated POST request to logout user sessions.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)}")