feat: add app user-profile schema tools

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`                        |
+| `get_app_user_schema`         | Read an app's user-profile schema (base + custom attributes) | - `What custom attributes does this app's profile have?` <br> - `Show the Profile Editor schema for the HR app`                                              |
+| `add_app_user_schema_attribute` | Add/update a custom attribute on an app's user-profile schema | - `Add an employeeNumber attribute to this app's profile` <br> - `Create a custom employmentStartDate field on the HR app`                                   |
 
 ### Policies
 
@@ -660,6 +662,8 @@ The Okta MCP Server uses a **scope-based tool loading** mechanism to ensure that
 | `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.schemas.read` | `get_app_user_schema` |
+| `okta.schemas.manage` | `add_app_user_schema_attribute` |
 | `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

@@ -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
@@ -444,3 +445,128 @@ 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 schema (Profile Editor)
+# ---------------------------------------------------------------------------
+
+
+@mcp.tool()
+@require_scopes("okta.schemas.read")
+@validate_ids("app_id", error_return_type="dict")
+async def get_app_user_schema(ctx: Context, app_id: str) -> Any:
+    """Get the default app user-profile schema for an application.
+
+    Returns the base and custom property definitions — the schema shown in Okta's
+    Profile Editor for the app. Use it to see which custom attributes already exist
+    before adding one.
+
+    Parameters:
+        app_id (str, required): The ID of the application
+
+    Returns:
+        Dict with the app user schema, or error information.
+    """
+    logger.info(f"Getting app user schema for application: {app_id}")
+
+    manager = ctx.request_context.lifespan_context.okta_auth_manager
+
+    try:
+        client = await get_okta_client(manager)
+        schema, _, err = await client.get_application_user_schema(app_id)
+
+        if err:
+            logger.error(f"Okta API error while getting app user schema for {app_id}: {err}")
+            return {"error": str(err)}
+
+        logger.info(f"Successfully retrieved app user schema for application: {app_id}")
+        return schema.to_dict() if schema else {}
+    except Exception as e:
+        logger.error(f"Exception while getting app user schema for {app_id}: {type(e).__name__}: {e}")
+        return {"error": str(e)}
+
+
+@mcp.tool()
+@require_scopes("okta.schemas.manage")
+@validate_ids("app_id", error_return_type="dict")
+async def add_app_user_schema_attribute(
+    ctx: Context,
+    app_id: str,
+    variable_name: str,
+    title: str,
+    attribute_type: str = "string",
+    description: Optional[str] = None,
+    attribute_definition: Optional[Dict[str, Any]] = None,
+) -> Any:
+    """Add or update a custom attribute on an application's default user-profile schema.
+
+    This is the API equivalent of the Profile Editor "Add Attribute" step. After
+    adding, for example, ``employeeNumber``, it can be referenced in a SAML
+    attribute statement as ``user.employeeNumber``.
+
+    The Okta schemas endpoint has no typed SDK wrapper, so this issues the request
+    directly. The post is a partial update: only the named property is added or
+    updated; existing custom properties are left intact.
+
+    Parameters:
+        app_id (str, required): The ID of the application
+        variable_name (str, required): The custom property key (e.g. ``employeeNumber``)
+        title (str, required): Human-readable display name
+        attribute_type (str, optional): JSON schema type — ``string`` (default),
+            ``integer``, ``number``, ``boolean``, or ``array``
+        description (str, optional): Description for the attribute
+        attribute_definition (dict, optional): Extra Okta schema property fields to
+            merge into the property, e.g.
+            ``{"maxLength": 50, "permissions": [{"principal": "SELF", "action": "READ_WRITE"}]}``
+
+    Returns:
+        Dict with the updated app user schema, or error information.
+    """
+    logger.info(f"Adding custom attribute '{variable_name}' to app user schema for application: {app_id}")
+
+    manager = ctx.request_context.lifespan_context.okta_auth_manager
+
+    try:
+        client = await get_okta_client(manager)
+
+        property_definition: Dict[str, Any] = {"title": title, "type": attribute_type}
+        if description:
+            property_definition["description"] = description
+        if attribute_definition:
+            property_definition.update(attribute_definition)
+
+        body = {
+            "definitions": {
+                "custom": {
+                    "id": "#custom",
+                    "type": "object",
+                    "properties": {variable_name: property_definition},
+                    "required": [],
+                }
+            }
+        }
+
+        executor = client.get_request_executor()
+        request, err = await executor.create_request(
+            method="POST",
+            url=f"/api/v1/meta/schemas/apps/{app_id}/default",
+            body=body,
+            headers={},
+            oauth=False,
+            keep_empty_params=True,
+        )
+        if err:
+            logger.error(f"Error building app user schema request for {app_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 schema for {app_id}: {err}")
+            return {"error": str(err)}
+
+        logger.info(f"Successfully added custom attribute '{variable_name}' for application: {app_id}")
+        return json.loads(response_body) if response_body else {}
+    except Exception as e:
+        logger.error(f"Exception while updating app user schema for {app_id}: {type(e).__name__}: {e}")
+        return {"error": str(e)}

src/okta_mcp_server/utils/scope_registry.py

@@ -60,6 +60,11 @@
     "activate_application":                 "okta.apps.manage",
     "deactivate_application":               "okta.apps.manage",
     # ------------------------------------------------------------------
+    # App user-profile schema  (src/okta_mcp_server/tools/applications/applications.py)
+    # ------------------------------------------------------------------
+    "get_app_user_schema":                  "okta.schemas.read",
+    "add_app_user_schema_attribute":        "okta.schemas.manage",
+    # ------------------------------------------------------------------
     # Policies  (src/okta_mcp_server/tools/policies/policies.py)
     # ------------------------------------------------------------------
     "list_policies":                        "okta.policies.read",

tests/test_app_user_schema.py

@@ -0,0 +1,132 @@
+# 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 schema tools (Profile Editor parity)."""
+
+from __future__ import annotations
+
+import json
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from okta_mcp_server.tools.applications.applications import (
+    add_app_user_schema_attribute,
+    get_app_user_schema,
+)
+
+
+APP_ID = "0oaSCHEMAAPP0001"
+
+
+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 TestGetAppUserSchema:
+    @pytest.mark.asyncio
+    @patch("okta_mcp_server.tools.applications.applications.get_okta_client")
+    async def test_returns_schema_dict(self, mock_get_client):
+        schema = MagicMock()
+        schema.to_dict.return_value = {"definitions": {"custom": {"properties": {}}}}
+        client = AsyncMock()
+        client.get_application_user_schema.return_value = (schema, None, None)
+        mock_get_client.return_value = client
+
+        result = await get_app_user_schema(ctx=_make_ctx(), app_id=APP_ID)
+
+        assert result == {"definitions": {"custom": {"properties": {}}}}
+
+    @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.get_application_user_schema.return_value = (None, None, "404 not found")
+        mock_get_client.return_value = client
+
+        result = await get_app_user_schema(ctx=_make_ctx(), app_id=APP_ID)
+
+        assert result == {"error": "404 not found"}
+
+
+def _client_posting(body, execute_error=None):
+    executor = MagicMock()
+    executor.create_request = AsyncMock(return_value=({"method": "POST"}, 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 TestAddAppUserSchemaAttribute:
+    @pytest.mark.asyncio
+    @patch("okta_mcp_server.tools.applications.applications.get_okta_client")
+    async def test_posts_custom_property_to_schema_endpoint(self, mock_get_client):
+        returned = {"definitions": {"custom": {"properties": {"employeeNumber": {"title": "Employee Number"}}}}}
+        client = _client_posting(json.dumps(returned))
+        mock_get_client.return_value = client
+
+        result = await add_app_user_schema_attribute(
+            ctx=_make_ctx(),
+            app_id=APP_ID,
+            variable_name="employeeNumber",
+            title="Employee Number",
+        )
+
+        assert result == returned
+
+        create_kwargs = client.get_request_executor.return_value.create_request.call_args.kwargs
+        assert create_kwargs["method"] == "POST"
+        assert create_kwargs["url"].endswith(f"/api/v1/meta/schemas/apps/{APP_ID}/default")
+        prop = create_kwargs["body"]["definitions"]["custom"]["properties"]["employeeNumber"]
+        assert prop == {"title": "Employee Number", "type": "string"}
+
+    @pytest.mark.asyncio
+    @patch("okta_mcp_server.tools.applications.applications.get_okta_client")
+    async def test_merges_attribute_definition_and_type_description(self, mock_get_client):
+        client = _client_posting(json.dumps({}))
+        mock_get_client.return_value = client
+
+        await add_app_user_schema_attribute(
+            ctx=_make_ctx(),
+            app_id=APP_ID,
+            variable_name="startDate",
+            title="Start Date",
+            attribute_type="string",
+            description="Employment start date",
+            attribute_definition={"permissions": [{"principal": "SELF", "action": "READ_WRITE"}]},
+        )
+
+        body = client.get_request_executor.return_value.create_request.call_args.kwargs["body"]
+        prop = body["definitions"]["custom"]["properties"]["startDate"]
+        assert prop["title"] == "Start Date"
+        assert prop["type"] == "string"
+        assert prop["description"] == "Employment start date"
+        assert prop["permissions"] == [{"principal": "SELF", "action": "READ_WRITE"}]
+
+    @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 = _client_posting(None, execute_error="403 forbidden")
+        mock_get_client.return_value = client
+
+        result = await add_app_user_schema_attribute(
+            ctx=_make_ctx(), app_id=APP_ID, variable_name="x", title="X"
+        )
+
+        assert result == {"error": "403 forbidden"}