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
7 changes: 7 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -499,6 +499,13 @@ The Okta MCP Server provides the following tools for LLMs to interact with your
| `activate_policy_rule` | Activate a policy rule | - `Activate the new emergency access rule` <br> - `Enable the contractor restrictions` <br> - `Turn on the location-based access rule` |
| `deactivate_policy_rule` | Deactivate a policy rule | - `Deactivate the old emergency rule` <br> - `Temporarily disable location restrictions` <br> - `Turn off the device trust requirements for testing` |

### Profile Mappings

| Tool | Description | Usage Examples |
| ------------------------- | ---------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `list_profile_mappings` | List profile mappings with optional source/target filter | - `Show me the profile mappings for the Salesforce application` <br> - `What mappings target the Okta user profile?` <br> - `List all mappings where this app is the source` |
| `get_profile_mapping` | Get a specific profile mapping with property expressions | - `Show me the attribute mapping details for this mapping` <br> - `What expression is used to map the department field?` <br> - `How is the email attribute transformed between the app and Okta?` |

### Logs

| Tool | Description | Usage Examples |
Expand Down
1 change: 1 addition & 0 deletions src/okta_mcp_server/server.py
Original file line number Diff line number Diff line change
Expand Up @@ -66,6 +66,7 @@ def main():
logger.info("Starting Okta MCP Server")
from okta_mcp_server.tools.applications import applications # noqa: F401
from okta_mcp_server.tools.groups import groups # noqa: F401
from okta_mcp_server.tools.mappings import mappings # noqa: F401
from okta_mcp_server.tools.policies import policies # noqa: F401
from okta_mcp_server.tools.system_logs import system_logs # noqa: F401
from okta_mcp_server.tools.users import users # noqa: F401
Expand Down
Empty file.
91 changes: 91 additions & 0 deletions src/okta_mcp_server/tools/mappings/mappings.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,91 @@
from typing import Any, Optional

from loguru import logger
from mcp.server.fastmcp import Context

from okta_mcp_server.server import mcp
from okta_mcp_server.utils.client import get_okta_client


@mcp.tool()
async def list_profile_mappings(
ctx: Context,
source_id: Optional[str] = None,
target_id: Optional[str] = None,
after: Optional[str] = None,
limit: Optional[int] = None,
) -> Any:
"""List profile mappings in the Okta organization.

Parameters:
source_id (str, optional): Filter by source ID (app or user type ID)
target_id (str, optional): Filter by target ID (app or user type ID)
after (str, optional): Pagination cursor for the next page of results
limit (int, optional): Maximum number of mappings to return

Returns:
List of profile mappings.
"""
logger.info("Listing profile mappings")
logger.debug(f"Query parameters: source_id={source_id}, target_id={target_id}")

manager = ctx.request_context.lifespan_context.okta_auth_manager

try:
client = await get_okta_client(manager)

query_params = {}
if source_id:
query_params["sourceId"] = source_id
if target_id:
query_params["targetId"] = target_id
if after:
query_params["after"] = after
if limit:
query_params["limit"] = limit

mappings, _, err = await client.list_profile_mappings(query_params)

if err:
logger.error(f"Okta API error while listing profile mappings: {err}")
return {"error": str(err)}

if not mappings:
logger.info("No profile mappings found")
return []

logger.info(f"Successfully retrieved {len(mappings)} profile mappings")
return [m for m in mappings]
except Exception as e:
logger.error(f"Exception while listing profile mappings: {type(e).__name__}: {e}")
return {"error": str(e)}


@mcp.tool()
async def get_profile_mapping(ctx: Context, mapping_id: str) -> Any:
"""Get a profile mapping by ID.

Parameters:
mapping_id (str, required): The ID of the profile mapping to retrieve

Returns:
Dictionary containing the profile mapping details including property mappings and expressions.
"""
logger.info(f"Getting profile mapping: {mapping_id}")

manager = ctx.request_context.lifespan_context.okta_auth_manager

try:
client = await get_okta_client(manager)

mapping, _, err = await client.get_profile_mapping(mapping_id)

if err:
logger.error(f"Okta API error while getting profile mapping {mapping_id}: {err}")
return {"error": str(err)}

logger.info(f"Successfully retrieved profile mapping: {mapping_id}")
return mapping
except Exception as e:
logger.error(f"Exception while getting profile mapping {mapping_id}: {type(e).__name__}: {e}")
return {"error": str(e)}