feat: add tools to assign users and groups to applications

README.md

@@ -487,6 +487,8 @@ The Okta MCP Server provides the following tools for LLMs to interact with your
 | `delete_application`          | Delete an application (prompts for confirmation)  | - `Delete the old legacy application` <br> - `Remove the unused test application` <br> - `Clean up deprecated integrations`                                   |
 | `activate_application`        | Activate an application                           | - `Activate the new HR application` <br> - `Enable the Salesforce integration` <br> - `Turn on the mobile app for users`                                      |
 | `deactivate_application`      | Deactivate an application (prompts for confirmation) | - `Deactivate the legacy CRM application` <br> - `Temporarily disable the mobile app` <br> - `Turn off access to the test environment`                        |
+| `assign_user_to_app`          | Assign a user to an application                   | - `Assign Jane to the Salesforce application` <br> - `Give this user access to the HR app`                                                                    |
+| `assign_group_to_app`         | Assign a group to an application                  | - `Assign the Engineering group to the GitHub app` <br> - `Give the Sales team access to Salesforce`                                                          |
 
 ### Policies
 
@@ -659,7 +661,7 @@ The Okta MCP Server uses a **scope-based tool loading** mechanism to ensure that
 | `okta.groups.read` | `list_groups`, `get_group`, `list_group_users`, `list_group_apps` |
 | `okta.groups.manage` | `create_group`, `update_group`, `delete_group`, `add_user_to_group`, `remove_user_from_group` |
 | `okta.apps.read` | `list_applications`, `get_application` |
-| `okta.apps.manage` | `create_application`, `update_application`, `delete_application`, `activate_application`, `deactivate_application` |
+| `okta.apps.manage` | `create_application`, `update_application`, `delete_application`, `activate_application`, `deactivate_application`, `assign_user_to_app`, `assign_group_to_app` |
 | `okta.policies.read` | `list_policies`, `get_policy`, `list_policy_rules`, `get_policy_rule` |
 | `okta.policies.manage` | `create_policy`, `update_policy`, `delete_policy`, `activate_policy`, `deactivate_policy`, `create_policy_rule`, `update_policy_rule`, `delete_policy_rule`, `activate_policy_rule`, `deactivate_policy_rule` |
 | `okta.deviceAssurance.read` | `list_device_assurance_policies`, `get_device_assurance_policy` |

src/okta_mcp_server/tools/applications/applications.py

@@ -444,3 +444,74 @@ async def deactivate_application(ctx: Context, app_id: str) -> list:
     except Exception as e:
         logger.error(f"Exception while deactivating application {app_id}: {type(e).__name__}: {e}")
         return [f"Exception: {e}"]
+
+
+@mcp.tool()
+@require_scopes("okta.apps.manage", error_return_type="list")
+@validate_ids("app_id", "user_id")
+async def assign_user_to_app(ctx: Context, app_id: str, user_id: str) -> list:
+    """Assign a user to an application in the Okta organization.
+
+    Parameters:
+        app_id (str, required): The ID of the application to assign the user to
+        user_id (str, required): The ID of the user to assign
+
+    Returns:
+        List containing the assigned application-user object, or error information.
+    """
+    logger.info(f"Assigning user {user_id} to application {app_id}")
+
+    manager = ctx.request_context.lifespan_context.okta_auth_manager
+
+    try:
+        client = await get_okta_client(manager)
+        app_user = okta_models.AppUserAssignRequest(id=user_id, scope="USER")
+        logger.debug(f"Calling Okta API to assign user {user_id} to application {app_id}")
+
+        assignment, _, err = await client.assign_user_to_application(app_id, app_user)
+
+        if err:
+            logger.error(f"Okta API error while assigning user {user_id} to application {app_id}: {err}")
+            return [f"Error: {err}"]
+
+        logger.info(f"Successfully assigned user {user_id} to application {app_id}")
+        return [assignment]
+    except Exception as e:
+        logger.error(f"Exception while assigning user {user_id} to application {app_id}: {type(e).__name__}: {e}")
+        return [f"Exception: {e}"]
+
+
+@mcp.tool()
+@require_scopes("okta.apps.manage", error_return_type="list")
+@validate_ids("app_id", "group_id")
+async def assign_group_to_app(ctx: Context, app_id: str, group_id: str) -> list:
+    """Assign a group to an application in the Okta organization.
+
+    All users in the group gain access to the application.
+
+    Parameters:
+        app_id (str, required): The ID of the application to assign the group to
+        group_id (str, required): The ID of the group to assign
+
+    Returns:
+        List containing the application-group assignment object, or error information.
+    """
+    logger.info(f"Assigning group {group_id} to application {app_id}")
+
+    manager = ctx.request_context.lifespan_context.okta_auth_manager
+
+    try:
+        client = await get_okta_client(manager)
+        logger.debug(f"Calling Okta API to assign group {group_id} to application {app_id}")
+
+        assignment, _, err = await client.assign_group_to_application(app_id, group_id)
+
+        if err:
+            logger.error(f"Okta API error while assigning group {group_id} to application {app_id}: {err}")
+            return [f"Error: {err}"]
+
+        logger.info(f"Successfully assigned group {group_id} to application {app_id}")
+        return [assignment]
+    except Exception as e:
+        logger.error(f"Exception while assigning group {group_id} to application {app_id}: {type(e).__name__}: {e}")
+        return [f"Exception: {e}"]

src/okta_mcp_server/utils/scope_registry.py

@@ -59,6 +59,8 @@
     "confirm_delete_application":           "okta.apps.manage",
     "activate_application":                 "okta.apps.manage",
     "deactivate_application":               "okta.apps.manage",
+    "assign_user_to_app":                   "okta.apps.manage",
+    "assign_group_to_app":                  "okta.apps.manage",
     # ------------------------------------------------------------------
     # Policies  (src/okta_mcp_server/tools/policies/policies.py)
     # ------------------------------------------------------------------

tests/test_app_assignment.py

@@ -0,0 +1,97 @@
+# The Okta software accompanied by this notice is provided pursuant to the following terms:
+# Copyright © 2026-Present, Okta, Inc.
+# Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0.
+# Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and limitations under the License.
+
+"""Tests for assign_user_to_app / assign_group_to_app."""
+
+from __future__ import annotations
+
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from okta.models import AppUserAssignRequest
+
+from okta_mcp_server.tools.applications.applications import assign_group_to_app, assign_user_to_app
+
+
+APP_ID = "0oaTESTAPP0000001"
+USER_ID = "00uTESTUSER000001"
+GROUP_ID = "00gTESTGROUP00001"
+
+
+def _make_ctx():
+    from tests.conftest import FakeLifespanContext, FakeOktaAuthManager
+
+    request_context = MagicMock()
+    request_context.lifespan_context = FakeLifespanContext(
+        okta_auth_manager=FakeOktaAuthManager()
+    )
+    ctx = MagicMock()
+    ctx.request_context = request_context
+    return ctx
+
+
+class TestAssignUserToApp:
+    @pytest.mark.asyncio
+    @patch("okta_mcp_server.tools.applications.applications.get_okta_client")
+    async def test_success_sends_assign_request_and_returns_assignment(self, mock_get_client):
+        app_user = MagicMock()
+        client = AsyncMock()
+        client.assign_user_to_application.return_value = (app_user, MagicMock(), None)
+        mock_get_client.return_value = client
+
+        result = await assign_user_to_app(ctx=_make_ctx(), app_id=APP_ID, user_id=USER_ID)
+
+        assert result == [app_user]
+        call_args = client.assign_user_to_application.call_args[0]
+        assert call_args[0] == APP_ID
+        assert isinstance(call_args[1], AppUserAssignRequest)
+        assert call_args[1].id == USER_ID
+        assert call_args[1].scope == "USER"
+
+    @pytest.mark.asyncio
+    @patch("okta_mcp_server.tools.applications.applications.get_okta_client")
+    async def test_api_error_is_returned(self, mock_get_client):
+        client = AsyncMock()
+        client.assign_user_to_application.return_value = (None, None, "404 user not found")
+        mock_get_client.return_value = client
+
+        result = await assign_user_to_app(ctx=_make_ctx(), app_id=APP_ID, user_id=USER_ID)
+
+        assert result == ["Error: 404 user not found"]
+
+    @pytest.mark.asyncio
+    async def test_invalid_id_rejected_before_api_call(self):
+        result = await assign_user_to_app(ctx=_make_ctx(), app_id="../../etc", user_id=USER_ID)
+
+        assert len(result) == 1
+        assert result[0].startswith("Error:")
+
+
+class TestAssignGroupToApp:
+    @pytest.mark.asyncio
+    @patch("okta_mcp_server.tools.applications.applications.get_okta_client")
+    async def test_success_returns_assignment(self, mock_get_client):
+        assignment = MagicMock()
+        client = AsyncMock()
+        client.assign_group_to_application.return_value = (assignment, MagicMock(), None)
+        mock_get_client.return_value = client
+
+        result = await assign_group_to_app(ctx=_make_ctx(), app_id=APP_ID, group_id=GROUP_ID)
+
+        assert result == [assignment]
+        client.assign_group_to_application.assert_awaited_once_with(APP_ID, GROUP_ID)
+
+    @pytest.mark.asyncio
+    @patch("okta_mcp_server.tools.applications.applications.get_okta_client")
+    async def test_api_error_is_returned(self, mock_get_client):
+        client = AsyncMock()
+        client.assign_group_to_application.return_value = (None, None, "403 forbidden")
+        mock_get_client.return_value = client
+
+        result = await assign_group_to_app(ctx=_make_ctx(), app_id=APP_ID, group_id=GROUP_ID)
+
+        assert result == ["Error: 403 forbidden"]