# src/fctr_okta_mcp/tools/okta/user_tools.py
"""
User Management Tools for Okta MCP Server.
Tools for listing, searching, and managing Okta users,
including their MFA factors, group memberships, and applications.
"""
from typing import Annotated
from fastmcp import FastMCP, Context
from fctr_okta_mcp.utils.logger import get_logger, log_audit_event
from fctr_okta_mcp.utils.response_compressor import compress_response
logger = get_logger(__name__)
def register_user_tools(mcp: FastMCP):
"""Register all user-related tools with the MCP server."""
logger.debug("Registering user tools...")
@mcp.tool(
tags={"okta", "user", "direct"},
annotations={
"category": "user",
"readOnlyHint": True,
"idempotentHint": True
}
)
async def okta_user_list(
ctx: Context,
search: Annotated[str, "SCIM filter syntax for advanced filtering. Searches profile attributes (prefix with 'profile.'), status, created, activated, etc. Example: 'profile.lastName eq \"Smith\" and status eq \"ACTIVE\"'. Supports: eq, sw, co (limited), gt, lt. Use 'and', 'or' for combining. Cannot combine with 'q' or 'filter'."] = "",
filter: Annotated[str, "SCIM filter for limited properties only: status, lastUpdated, id, profile.login, profile.email, profile.firstName, profile.lastName. Example: 'status eq \"LOCKED_OUT\"'. Cannot combine with 'search'."] = "",
q: Annotated[str, "Simple text search matching firstName, lastName, or email (startsWith match). Note: Omits DEPROVISIONED users. Cannot combine with 'search'."] = "",
after: Annotated[str, "Pagination cursor from previous response's _links.next for fetching next page."] = "",
sortBy: Annotated[str, "Field to sort by (search queries only). Example: 'profile.lastName'."] = "",
sortOrder: Annotated[str, "Sort direction: 'asc' or 'desc' (search queries only)."] = "",
expand: Annotated[str, "Use 'classification' to include user classification metadata in _embedded property."] = ""
) -> dict:
"""
IMPORTANT: Call read_system_instructions() FIRST before using this tool!
Returns SAMPLE DATA (3 results) + endpoint metadata for code generation.
This is NOT the final answer - you must generate Python code for full results.
List users in your Okta organization.
## User Status Values
- STAGED: User is staged but not yet activated
- PROVISIONED: User is provisioned but not activated
- ACTIVE: User is active and can authenticate
- PASSWORD_RESET: User password requires reset
- PASSWORD_EXPIRED: User password has expired
- LOCKED_OUT: User account is locked due to failed login attempts
- SUSPENDED: User account is temporarily suspended
- DEPROVISIONED: User account has been deprovisioned/deleted
## Parameter Usage Notes
- CRITICAL: Use 'search' for advanced SCIM filtering, 'filter' for limited system properties, 'q' for simple name/email matching
- Mixing search and filter parameters causes errors
- The 'q' parameter automatically excludes DEPROVISIONED users
- Search queries can be sorted with sortBy/sortOrder
- To find DEPROVISIONED users, use: filter=status eq "DEPROVISIONED"
## Search Examples
- Status: search=status eq "ACTIVE"
- Department: search=profile.department eq "Engineering"
- Time-based: search=lastUpdated gt "2014-01-01T00:00:00.000Z"
- Complex: search=profile.department eq "Engineering" and status eq "ACTIVE"
- Exclude status: search=(status lt "STAGED" or status gt "STAGED")
## Important Notes
- The 'ne' (not equal) operator is NOT supported; use lt/gt combinations instead
- All SCIM expressions must be URL encoded
- Property names are case sensitive, operators and values are case insensitive
"""
# Build filter description for logging
filter_desc = []
if search:
filter_desc.append(f"search='{search}'")
if filter:
filter_desc.append(f"filter='{filter}'")
if q:
filter_desc.append(f"q='{q}'")
filter_log = ", ".join(filter_desc) if filter_desc else "no filters"
await ctx.info(f"Fetching sample users ({filter_log})")
from fctr_okta_mcp.client.base_okta_api_client import OktaAPIClient
client = OktaAPIClient(ctx=ctx)
params = {"limit": 3} # Hardcoded limit of 3 for sample data
if search:
params["search"] = search
if filter:
params["filter"] = filter
if q:
params["q"] = q
if after:
params["after"] = after
if sortBy:
params["sortBy"] = sortBy
if sortOrder:
params["sortOrder"] = sortOrder
if expand:
params["expand"] = expand
await ctx.info("Calling Okta API: GET /api/v1/users...")
result = await client.make_request(
method="GET",
endpoint="/api/v1/users",
params=params,
max_results=3 # Stop after 3 results total
)
# Add reminder in the response
if result.get("status") == "success":
data = result.get("data", [])
await ctx.info(f"Retrieved {len(data)} sample user(s)")
result["_reminder"] = "This is SAMPLE DATA (3 results). Generate Python code using execute_code() to get full results."
log_audit_event(
action="okta_user_list",
status="success",
details={"filter": filter_log, "result_count": len(data)}
)
else:
await ctx.warning(f"API returned status: {result.get('status')}")
log_audit_event(
action="okta_user_list",
status="error",
details={"filter": filter_log, "api_status": result.get("status")}
)
return compress_response(result)
@mcp.tool(
tags={"okta", "user", "authenticator", "direct"},
annotations={
"category": "user",
"readOnlyHint": True,
"idempotentHint": True
}
)
async def okta_user_list_factors(
ctx: Context,
userId: Annotated[str, "The ID of the user"]
) -> dict:
"""
**CALL read_system_instructions() FIRST!**
Lists all MFA factors enrolled by the specified user.
## Factor Type Values
- **sms** - SMS text message authentication
- **email** - Email-based authentication
- **signed_nonce** - FastPass/Okta FastPass (passwordless)
- **password** - Password factor
- **webauthn** - FIDO2 security keys (WebAuthn standard)
- **security_question** - Security question challenge
- **token:software:totp** - Time-based one-time password (TOTP) apps like Google Authenticator, Authy
- **push** - Okta Verify push notifications
## Authenticator Name Examples
Human-readable names include: "Google Authenticator", "Okta Verify", "YubiKey", "Duo Security", "SMS Authentication"
## Important Notes
- This is the primary endpoint to see which MFA factors a user has set up.
- **Note**: This is a legacy endpoint. For new development, use list-user-authenticator-enrollments for more detailed representation.
- Factor response includes: factorType, provider, status, authenticator_name (human-readable), and device details (for mobile/hardware factors).
Returns SAMPLE DATA (3 results) + endpoint metadata for code generation.
You MUST generate Python code to get actual results - see read_system_instructions().
"""
await ctx.info(f"Fetching sample MFA factors for user: {userId}")
from fctr_okta_mcp.client.base_okta_api_client import OktaAPIClient
client = OktaAPIClient(ctx=ctx)
await ctx.info(f"Calling Okta API: GET /api/v1/users/{userId}/factors...")
result = await client.make_request(
method="GET",
endpoint=f"/api/v1/users/{userId}/factors",
params={},
max_results=3 # Stop after 3 results total
)
# Add reminder in the response
if result.get("status") == "success":
data = result.get("data", [])
factor_types = [f.get("factorType", "unknown") for f in data]
await ctx.info(f"Retrieved {len(data)} factor(s): {', '.join(factor_types) if factor_types else 'none'}")
result["_reminder"] = "This is SAMPLE DATA (3 results). Generate Python code using execute_code() to get full results."
log_audit_event(
action="okta_user_list_factors",
status="success",
details={"user_id": userId, "result_count": len(data)}
)
else:
await ctx.warning(f"API returned status: {result.get('status')}")
log_audit_event(
action="okta_user_list_factors",
status="error",
details={"user_id": userId, "api_status": result.get("status")}
)
return compress_response(result)
@mcp.tool(
tags={"okta", "user", "group", "direct"},
annotations={
"category": "user",
"readOnlyHint": True,
"idempotentHint": True
}
)
async def okta_user_list_groups(
ctx: Context,
userId: Annotated[str, "The ID of the user (required)"]
) -> dict:
"""
**CALL read_system_instructions() FIRST!**
Lists all groups of which the user is a member.
## Path Parameters
- **userId** (String, Required): The ID of the user.
## Response Fields
- **id**: Group ID
- **profile.name**: Group name
- **profile.description**: Group description
- **type**: Group type (OKTA_GROUP, APP_GROUP, BUILT_IN)
## Important Notes
- Groups provide access to applications through group assignments.
- For complete access analysis, also check user's appLinks and roles.
- Use this with okta_user_list to get groups for multiple users.
## Common Use Cases
- Audit user's group memberships
- Analyze access through group assignments
- Compare group memberships across users
Returns SAMPLE DATA (3 results) + endpoint metadata for code generation.
You MUST generate Python code to get actual results - see read_system_instructions().
"""
await ctx.info(f"Fetching sample group memberships for user: {userId}")
from fctr_okta_mcp.client.base_okta_api_client import OktaAPIClient
client = OktaAPIClient(ctx=ctx)
await ctx.info(f"Calling Okta API: GET /api/v1/users/{userId}/groups...")
result = await client.make_request(
method="GET",
endpoint=f"/api/v1/users/{userId}/groups",
params={},
max_results=3
)
# Add result handling
if result.get("status") == "success":
data = result.get("data", [])
group_names = [g.get("profile", {}).get("name", "unknown") for g in data]
await ctx.info(f"Retrieved {len(data)} group(s): {', '.join(group_names) if group_names else 'none'}")
result["_reminder"] = "This is SAMPLE DATA (3 results). Generate Python code using execute_code() to get full results."
log_audit_event(
action="okta_user_list_groups",
status="success",
details={"user_id": userId, "result_count": len(data)}
)
else:
await ctx.warning(f"API returned status: {result.get('status')}")
log_audit_event(
action="okta_user_list_groups",
status="error",
details={"user_id": userId, "api_status": result.get("status")}
)
return compress_response(result)
@mcp.tool(
tags={"okta", "user", "direct"},
annotations={
"category": "user",
"readOnlyHint": True,
"idempotentHint": True
}
)
async def okta_user_get(
ctx: Context,
userId: Annotated[str, "The ID or login (email) of the user to retrieve"]
) -> dict:
"""
**CALL read_system_instructions() FIRST!**
Get detailed information about a specific user by their ID or login email.
## Path Parameters
- **userId** (String, Required): User ID (e.g., '00ub0oNGTSWTBKOLGLNR') or user login (email).
## Response Fields
- **id**: User's unique identifier
- **status**: User status (STAGED, PROVISIONED, ACTIVE, etc.)
- **created**: When the user was created
- **activated**: When the user was activated
- **lastLogin**: User's last successful login
- **lastUpdated**: When the user profile was last modified
- **passwordChanged**: When the password was last changed
- **profile**: Contains firstName, lastName, email, login, and custom attributes
- **credentials**: Authentication credentials info (provider, recovery question)
## User Status Values
- **STAGED**: User created but not activated
- **PROVISIONED**: User provisioned but hasn't logged in
- **ACTIVE**: User can authenticate
- **PASSWORD_RESET**: Password requires reset
- **PASSWORD_EXPIRED**: Password has expired
- **LOCKED_OUT**: Account locked due to failed attempts
- **SUSPENDED**: Account temporarily suspended
- **DEPROVISIONED**: Account deleted/deprovisioned
## Common Use Cases
- Get full user profile details
- Check user status before operations
- Verify user exists by email/login
- Audit user account information
"""
await ctx.info(f"Fetching user details: {userId}")
from fctr_okta_mcp.client.base_okta_api_client import OktaAPIClient
client = OktaAPIClient(ctx=ctx)
await ctx.info(f"Calling Okta API: GET /api/v1/users/{userId}...")
result = await client.make_request(
method="GET",
endpoint=f"/api/v1/users/{userId}",
params={}
)
if result.get("status") == "success":
data = result.get("data", {})
user_email = data.get("profile", {}).get("email", "unknown")
user_status = data.get("status", "unknown")
await ctx.info(f"Retrieved user: {user_email} (status: {user_status})")
log_audit_event(
action="okta_user_get",
status="success",
details={"user_id": userId, "email": user_email, "status": user_status}
)
else:
await ctx.warning(f"API returned status: {result.get('status')}")
log_audit_event(
action="okta_user_get",
status="error",
details={"user_id": userId, "api_status": result.get("status")}
)
return compress_response(result)
@mcp.tool(
tags={"okta", "user", "application", "direct"},
annotations={
"category": "user",
"readOnlyHint": True,
"idempotentHint": True
}
)
async def okta_user_list_applications(
ctx: Context,
userId: Annotated[str, "The ID of the user (required)"],
showAll: Annotated[bool, "If True, returns all assigned app links even inactive ones. Default is False (active only)."] = False
) -> dict:
"""
**CALL read_system_instructions() FIRST!**
Lists all application links (appLinks) assigned to the user.
Shows which applications the user has access to through direct or group assignments.
## Path Parameters
- **userId** (String, Required): The ID of the user.
## Query Parameters
- **showAll** (Boolean, Optional): When True, includes inactive app links. Default is False.
## Response Fields
- **id**: AppLink ID
- **label**: Application display name
- **linkUrl**: URL to launch the application
- **logoUrl**: Application logo URL
- **appName**: Application name/type
- **appInstanceId**: Application instance ID
- **appAssignmentId**: Assignment identifier
- **credentialsSetup**: Whether credentials are configured
- **hidden**: Whether app is hidden from user dashboard
- **sortOrder**: Display order on user dashboard
## Common Use Cases
- Audit user's application access
- Generate user access reports
- Verify application assignments
- Compare access across users
Returns SAMPLE DATA (3 results) + endpoint metadata for code generation.
You MUST generate Python code to get actual results - see read_system_instructions().
"""
await ctx.info(f"Fetching application links for user: {userId}")
from fctr_okta_mcp.client.base_okta_api_client import OktaAPIClient
client = OktaAPIClient(ctx=ctx)
params = {}
if showAll:
params["showAll"] = "true"
await ctx.info(f"Calling Okta API: GET /api/v1/users/{userId}/appLinks...")
result = await client.make_request(
method="GET",
endpoint=f"/api/v1/users/{userId}/appLinks",
params=params,
max_results=3
)
if result.get("status") == "success":
data = result.get("data", [])
app_labels = [a.get("label", "unknown") for a in data]
await ctx.info(f"Retrieved {len(data)} app link(s): {', '.join(app_labels) if app_labels else 'none'}")
result["_reminder"] = "This is SAMPLE DATA (3 results). Generate Python code using execute_code() to get full results."
log_audit_event(
action="okta_user_list_applications",
status="success",
details={"user_id": userId, "result_count": len(data)}
)
else:
await ctx.warning(f"API returned status: {result.get('status')}")
log_audit_event(
action="okta_user_list_applications",
status="error",
details={"user_id": userId, "api_status": result.get("status")}
)
return compress_response(result)
logger.info("User tools registered successfully")
