Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 4 additions & 2 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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` |
| `get_app_user` | Read a user's app assignment, including their app-specific profile | - `What's Jane's profile on the HR app?` <br> - `Show this user's app assignment status` |
| `update_app_user_profile` | Set a user's app-specific profile attribute values | - `Set employeeNumber to E-12345 for Jane on the HR app` <br> - `Populate this user's app profile values` |

### Policies

Expand Down Expand Up @@ -658,8 +660,8 @@ The Okta MCP Server uses a **scope-based tool loading** mechanism to ensure that
| `okta.users.manage` | `create_user`, `update_user`, `deactivate_user`, `delete_deactivated_user` |
| `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.read` | `list_applications`, `get_application`, `get_app_user` |
| `okta.apps.manage` | `create_application`, `update_application`, `delete_application`, `activate_application`, `deactivate_application`, `update_app_user_profile` |
| `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` |
Expand Down
98 changes: 98 additions & 0 deletions src/okta_mcp_server/tools/applications/applications.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
# 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.

import json
from typing import Any, Dict, Optional

import okta.models as okta_models
Expand Down Expand Up @@ -444,3 +445,100 @@ 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}"]


# ---------------------------------------------------------------------------
# App-user profile (per-assignment attribute values)
# ---------------------------------------------------------------------------


@mcp.tool()
@require_scopes("okta.apps.read")
@validate_ids("app_id", "user_id", error_return_type="dict")
async def get_app_user(ctx: Context, app_id: str, user_id: str) -> Any:
"""Get a user's assignment on an application, including their app-specific profile.

Shows the values of the app's user-profile attributes for this user (e.g.
``employeeNumber``), plus the assignment scope and status.

Parameters:
app_id (str, required): The ID of the application
user_id (str, required): The ID of the assigned user

Returns:
Dict with the application-user assignment (incl. profile), or error information.
"""
logger.info(f"Getting app-user {user_id} for application: {app_id}")

manager = ctx.request_context.lifespan_context.okta_auth_manager

try:
client = await get_okta_client(manager)
executor = client.get_request_executor()
request, err = await executor.create_request(
method="GET", url=f"/api/v1/apps/{app_id}/users/{user_id}", body={}, headers={}, oauth=False
)
if err:
logger.error(f"Error building get-app-user request for {app_id}/{user_id}: {err}")
return {"error": str(err)}

_, response_body, err = await executor.execute(request)
if err:
logger.error(f"Okta API error while getting app-user {user_id} for {app_id}: {err}")
return {"error": str(err)}

logger.info(f"Successfully retrieved app-user {user_id} for application: {app_id}")
return json.loads(response_body) if response_body else {}
except Exception as e:
logger.error(f"Exception while getting app-user {user_id} for {app_id}: {type(e).__name__}: {e}")
return {"error": str(e)}


@mcp.tool()
@require_scopes("okta.apps.manage")
@validate_ids("app_id", "user_id", error_return_type="dict")
async def update_app_user_profile(ctx: Context, app_id: str, user_id: str, profile: Dict[str, Any]) -> Any:
"""Set a user's app-specific profile attribute values on an application.

Use this to populate the per-user values of attributes defined on the app's
user-profile schema — for example ``{"employeeNumber": "E-12345"}`` after the
``employeeNumber`` attribute has been added with add_app_user_schema_attribute.
The user must already be assigned to the application.

Parameters:
app_id (str, required): The ID of the application
user_id (str, required): The ID of the assigned user
profile (dict, required): App-user profile attribute values to set, e.g.
``{"employeeNumber": "E-12345", "employmentStartDate": "2026-01-01"}``

Returns:
Dict with the updated application-user assignment, or error information.
"""
logger.info(f"Updating app-user profile for {user_id} on application: {app_id}")

manager = ctx.request_context.lifespan_context.okta_auth_manager

try:
client = await get_okta_client(manager)
executor = client.get_request_executor()
request, err = await executor.create_request(
method="POST",
url=f"/api/v1/apps/{app_id}/users/{user_id}",
body={"profile": profile},
headers={},
oauth=False,
)
if err:
logger.error(f"Error building update-app-user request for {app_id}/{user_id}: {err}")
return {"error": str(err)}

_, response_body, err = await executor.execute(request)
if err:
logger.error(f"Okta API error while updating app-user {user_id} for {app_id}: {err}")
return {"error": str(err)}

logger.info(f"Successfully updated app-user profile for {user_id} on application: {app_id}")
return json.loads(response_body) if response_body else {}
except Exception as e:
logger.error(f"Exception while updating app-user {user_id} for {app_id}: {type(e).__name__}: {e}")
return {"error": str(e)}
2 changes: 2 additions & 0 deletions src/okta_mcp_server/utils/scope_registry.py
Original file line number Diff line number Diff line change
Expand Up @@ -59,6 +59,8 @@
"confirm_delete_application": "okta.apps.manage",
"activate_application": "okta.apps.manage",
"deactivate_application": "okta.apps.manage",
"get_app_user": "okta.apps.read",
"update_app_user_profile": "okta.apps.manage",
# ------------------------------------------------------------------
# Policies (src/okta_mcp_server/tools/policies/policies.py)
# ------------------------------------------------------------------
Expand Down
112 changes: 112 additions & 0 deletions tests/test_app_user_profile.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,112 @@
# 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 app-user profile tools (per-assignment attribute values)."""

from __future__ import annotations

import json
from unittest.mock import AsyncMock, MagicMock, patch

import pytest

from okta_mcp_server.tools.applications.applications import get_app_user, update_app_user_profile


APP_ID = "0oaHRAPP00000001"
USER_ID = "00uJANE000000001"


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


def _client_returning(body, execute_error=None):
executor = MagicMock()
executor.create_request = AsyncMock(return_value=({"method": "X"}, None))
if execute_error is not None:
executor.execute = AsyncMock(return_value=(None, None, execute_error))
else:
executor.execute = AsyncMock(return_value=(MagicMock(), body, None))
client = MagicMock()
client.get_request_executor = MagicMock(return_value=executor)
return client


class TestGetAppUser:
@pytest.mark.asyncio
@patch("okta_mcp_server.tools.applications.applications.get_okta_client")
async def test_returns_app_user_json(self, mock_get_client):
app_user = {"id": USER_ID, "scope": "USER", "status": "ACTIVE",
"profile": {"employeeNumber": "E-12345"}}
client = _client_returning(json.dumps(app_user))
mock_get_client.return_value = client

result = await get_app_user(ctx=_make_ctx(), app_id=APP_ID, user_id=USER_ID)

assert result["id"] == USER_ID
assert result["profile"]["employeeNumber"] == "E-12345"
kwargs = client.get_request_executor.return_value.create_request.call_args.kwargs
assert kwargs["method"] == "GET"
assert kwargs["url"].endswith(f"/api/v1/apps/{APP_ID}/users/{USER_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):
mock_get_client.return_value = _client_returning(None, execute_error="404 not found")

result = await get_app_user(ctx=_make_ctx(), app_id=APP_ID, user_id=USER_ID)

assert result == {"error": "404 not found"}


class TestUpdateAppUserProfile:
@pytest.mark.asyncio
@patch("okta_mcp_server.tools.applications.applications.get_okta_client")
async def test_posts_profile_to_app_user_endpoint(self, mock_get_client):
returned = {"id": USER_ID, "profile": {"employeeNumber": "E-12345"}}
client = _client_returning(json.dumps(returned))
mock_get_client.return_value = client

profile = {"employeeNumber": "E-12345", "employmentStartDate": "2026-01-01"}
result = await update_app_user_profile(
ctx=_make_ctx(), app_id=APP_ID, user_id=USER_ID, profile=profile
)

assert result == returned
kwargs = client.get_request_executor.return_value.create_request.call_args.kwargs
assert kwargs["method"] == "POST"
assert kwargs["url"].endswith(f"/api/v1/apps/{APP_ID}/users/{USER_ID}")
assert kwargs["body"] == {"profile": profile}

@pytest.mark.asyncio
@patch("okta_mcp_server.tools.applications.applications.get_okta_client")
async def test_api_error_is_returned(self, mock_get_client):
mock_get_client.return_value = _client_returning(None, execute_error="400 bad profile")

result = await update_app_user_profile(
ctx=_make_ctx(), app_id=APP_ID, user_id=USER_ID, profile={"x": 1}
)

assert result == {"error": "400 bad profile"}

@pytest.mark.asyncio
async def test_invalid_id_rejected_before_api_call(self):
result = await update_app_user_profile(
ctx=_make_ctx(), app_id=APP_ID, user_id="../../etc", profile={"x": 1}
)

assert isinstance(result, dict)
assert "error" in result