mcp-keycloak

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 message

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)}")