# src/fctr_okta_mcp/tools/okta/app_tools.py
"""
Application Management Tools for Okta MCP Server.
Tools for listing, searching, and managing Okta applications,
including application users and group assignments.
"""
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_app_tools(mcp: FastMCP):
"""Register all application-related tools with the MCP server."""
logger.debug("Registering application tools...")
@mcp.tool(
tags={"okta", "application", "direct"},
annotations={
"category": "application",
"readOnlyHint": True,
"idempotentHint": True
}
)
async def okta_app_list(
ctx: Context,
q: Annotated[str, "Search query to filter applications by name (startsWith match on label). Example: 'Salesforce'."] = "",
filter: Annotated[str, "SCIM filter expression for advanced filtering. Supported fields: status, user.id, group.id, credentials.signing.kid. Examples: 'status eq \"ACTIVE\"', 'user.id eq \"00ub0oNGTSWTBKOLGLNR\"'."] = "",
after: Annotated[str, "Pagination cursor from previous response's _links.next for fetching next page."] = "",
limit: Annotated[int, "Number of results per page (max: 200). Default is 20."] = 0,
expand: Annotated[str, "Include additional data. Values: 'user/{userId}' to embed application assignment for specific user."] = "",
includeNonDeleted: Annotated[bool, "If True, includes soft-deleted applications. Default is False."] = False
) -> 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.
Lists all applications in the organization with optional filtering.
## Query Parameters
- **q**: Simple search matching application label (startsWith).
- **filter**: SCIM filter for advanced filtering on status, user.id, group.id, credentials.signing.kid.
- **limit**: Results per page (max: 200, default: 20).
- **after**: Pagination cursor for next page.
- **expand**: Embed additional data like user assignment status.
- **includeNonDeleted**: Include soft-deleted applications.
## Filter Examples
- Active apps: `filter=status eq "ACTIVE"`
- Inactive apps: `filter=status eq "INACTIVE"`
- Apps assigned to user: `filter=user.id eq "00ub0oNGTSWTBKOLGLNR"`
- Apps assigned to group: `filter=group.id eq "00g1emaKYZTWRYYRRTSK"`
- Apps with specific signing key: `filter=credentials.signing.kid eq "SIMcCQNY3uwXoW3y0vf6VxiBb5n9pf8L2fK8d-FIbm4"`
## Response Fields
- **id**: Application's unique identifier
- **name**: Application type name (e.g., 'oidc_client', 'saml_2_0')
- **label**: Human-readable application name
- **status**: Application status (ACTIVE, INACTIVE)
- **created**: When application was created
- **lastUpdated**: When application was last modified
- **signOnMode**: Authentication mode (SAML_2_0, OPENID_CONNECT, etc.)
- **features**: List of enabled features
- **visibility**: Visibility settings
- **credentials**: Authentication credentials configuration
## Application Status Values
- **ACTIVE**: Application is active and accessible
- **INACTIVE**: Application is disabled
- **DELETED**: Application is soft-deleted (with includeNonDeleted)
## Common Use Cases
- Inventory all applications
- Find apps by name
- Audit inactive applications
- Find apps assigned to specific user/group
"""
filter_desc = []
if q:
filter_desc.append(f"q='{q}'")
if filter:
filter_desc.append(f"filter='{filter}'")
filter_log = ", ".join(filter_desc) if filter_desc else "no filters"
await ctx.info(f"Fetching sample applications ({filter_log})")
from fctr_okta_mcp.client.base_okta_api_client import OktaAPIClient
client = OktaAPIClient(ctx=ctx)
params = {"limit": 3} # Hardcoded for sample data
if q:
params["q"] = q
if filter:
params["filter"] = filter
if after:
params["after"] = after
if limit and limit > 0:
params["limit"] = min(limit, 3)
if expand:
params["expand"] = expand
if includeNonDeleted:
params["includeNonDeleted"] = "true"
await ctx.info("Calling Okta API: GET /api/v1/apps...")
result = await client.make_request(
method="GET",
endpoint="/api/v1/apps",
params=params,
max_results=3
)
if result.get("status") == "success":
data = result.get("data", [])
await ctx.info(f"Retrieved {len(data)} sample application(s)")
result["_reminder"] = "This is SAMPLE DATA (3 results). Generate Python code using execute_code() to get full results."
log_audit_event(
action="okta_app_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_app_list",
status="error",
details={"filter": filter_log, "api_status": result.get("status")}
)
return compress_response(result)
@mcp.tool(
tags={"okta", "application", "direct"},
annotations={
"category": "application",
"readOnlyHint": True,
"idempotentHint": True
}
)
async def okta_app_get(
ctx: Context,
appId: Annotated[str, "The ID of the application to retrieve (required)"],
expand: Annotated[str, "Include additional data. Value: 'user/{userId}' to embed assignment for specific user."] = ""
) -> dict:
"""
**CALL read_system_instructions() FIRST!**
Get detailed information about a specific application by its ID.
## Path Parameters
- **appId** (String, Required): The unique ID of the application.
## Query Parameters
- **expand** (String, Optional): Embed user assignment with 'user/{userId}'.
## Response Fields
- **id**: Application's unique identifier
- **name**: Application type name
- **label**: Human-readable application name
- **status**: Application status (ACTIVE, INACTIVE)
- **created**: When application was created
- **lastUpdated**: When application was last modified
- **signOnMode**: Authentication mode (SAML_2_0, OPENID_CONNECT, etc.)
- **features**: List of enabled features
- **settings**: Application-specific settings
- **credentials**: Authentication credentials configuration
- **visibility**: Visibility settings (auto-submit, hide links)
- **accessibility**: Accessibility settings
## SignOn Modes
- **BOOKMARK**: Simple bookmark app
- **BASIC_AUTH**: Basic authentication
- **BROWSER_PLUGIN**: Browser plugin-based auth
- **SECURE_PASSWORD_STORE**: Stored password app
- **SAML_2_0**: SAML 2.0 federation
- **WS_FEDERATION**: WS-Federation
- **OPENID_CONNECT**: OIDC authentication
- **AUTO_LOGIN**: Automatic login
## Common Use Cases
- Get application details before modifications
- Verify application configuration
- Review authentication settings
- Check credentials configuration
"""
await ctx.info(f"Fetching application details: {appId}")
from fctr_okta_mcp.client.base_okta_api_client import OktaAPIClient
client = OktaAPIClient(ctx=ctx)
params = {}
if expand:
params["expand"] = expand
await ctx.info(f"Calling Okta API: GET /api/v1/apps/{appId}...")
result = await client.make_request(
method="GET",
endpoint=f"/api/v1/apps/{appId}",
params=params
)
if result.get("status") == "success":
data = result.get("data", {})
app_label = data.get("label", "unknown")
app_status = data.get("status", "unknown")
await ctx.info(f"Retrieved application: {app_label} (status: {app_status})")
log_audit_event(
action="okta_app_get",
status="success",
details={"app_id": appId, "label": app_label, "status": app_status}
)
else:
await ctx.warning(f"API returned status: {result.get('status')}")
log_audit_event(
action="okta_app_get",
status="error",
details={"app_id": appId, "api_status": result.get("status")}
)
return compress_response(result)
@mcp.tool(
tags={"okta", "application", "user", "direct"},
annotations={
"category": "application",
"readOnlyHint": True,
"idempotentHint": True
}
)
async def okta_app_list_users(
ctx: Context,
appId: Annotated[str, "The ID of the application (required)"],
q: Annotated[str, "Search query to filter by user profile (firstName, lastName, email). Matches substring."] = "",
after: Annotated[str, "Pagination cursor from previous response for fetching next page."] = "",
limit: Annotated[int, "Number of results per page (max: 500). Default is 50."] = 0,
expand: Annotated[str, "Include additional data. Value: 'user' to embed full user profile."] = ""
) -> dict:
"""
**CALL read_system_instructions() FIRST!**
Lists all users assigned to the specified application.
## Path Parameters
- **appId** (String, Required): The ID of the application.
## Query Parameters
- **q** (String, Optional): Search user profiles (firstName, lastName, email).
- **after** (String, Optional): Pagination cursor.
- **limit** (Integer, Optional): Results per page (max: 500, default: 50).
- **expand** (String, Optional): Use 'user' to embed full user profile.
## Response Fields
Returns array of application user objects:
- **id**: Assignment ID
- **externalId**: External user ID (if applicable)
- **created**: When user was assigned
- **lastUpdated**: When assignment was last modified
- **scope**: Assignment scope (USER or GROUP)
- **status**: Assignment status (ACTIVE, INACTIVE, etc.)
- **statusChanged**: When status last changed
- **passwordChanged**: When app password last changed
- **syncState**: Provisioning sync state
- **credentials**: App-specific credentials
- **profile**: App-specific user profile
- **_embedded.user**: Full user profile (if expand=user)
## Assignment Status Values
- **ACTIVE**: User can access the app
- **INACTIVE**: Assignment is disabled
- **PENDING**: Awaiting activation
- **PROVISIONED**: Provisioned but not activated
## Common Use Cases
- List all users with access to an application
- Audit application assignments
- Find users by name in an application
- Export application user list
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 users assigned to application: {appId}")
from fctr_okta_mcp.client.base_okta_api_client import OktaAPIClient
client = OktaAPIClient(ctx=ctx)
params = {"limit": 3} # Hardcoded for sample data
if q:
params["q"] = q
if after:
params["after"] = after
if limit and limit > 0:
params["limit"] = min(limit, 3)
if expand:
params["expand"] = expand
await ctx.info(f"Calling Okta API: GET /api/v1/apps/{appId}/users...")
result = await client.make_request(
method="GET",
endpoint=f"/api/v1/apps/{appId}/users",
params=params,
max_results=3
)
if result.get("status") == "success":
data = result.get("data", [])
await ctx.info(f"Retrieved {len(data)} user assignment(s)")
result["_reminder"] = "This is SAMPLE DATA (3 results). Generate Python code using execute_code() to get full results."
log_audit_event(
action="okta_app_list_users",
status="success",
details={"app_id": appId, "result_count": len(data)}
)
else:
await ctx.warning(f"API returned status: {result.get('status')}")
log_audit_event(
action="okta_app_list_users",
status="error",
details={"app_id": appId, "api_status": result.get("status")}
)
return compress_response(result)
@mcp.tool(
tags={"okta", "application", "group", "direct"},
annotations={
"category": "application",
"readOnlyHint": True,
"idempotentHint": True
}
)
async def okta_app_list_groups(
ctx: Context,
appId: Annotated[str, "The ID of the application (required)"],
q: Annotated[str, "Search query to filter by group name (startsWith match)."] = "",
after: Annotated[str, "Pagination cursor from previous response for fetching next page."] = "",
limit: Annotated[int, "Number of results per page (max: 200). Default is 20."] = 0,
expand: Annotated[str, "Include additional data. Value: 'group' to embed full group profile."] = ""
) -> dict:
"""
**CALL read_system_instructions() FIRST!**
Lists all groups assigned to the specified application.
## Path Parameters
- **appId** (String, Required): The ID of the application.
## Query Parameters
- **q** (String, Optional): Search groups by name (startsWith).
- **after** (String, Optional): Pagination cursor.
- **limit** (Integer, Optional): Results per page (max: 200, default: 20).
- **expand** (String, Optional): Use 'group' to embed full group profile.
## Response Fields
Returns array of application group assignment objects:
- **id**: Group ID
- **lastUpdated**: When assignment was last modified
- **priority**: Assignment priority for licensing
- **profile**: App-specific group profile settings
- **_embedded.group**: Full group profile (if expand=group)
## Common Use Cases
- List all groups with access to an application
- Audit group-based application assignments
- Find groups by name assigned to an app
- Review application access governance
- Impact analysis before group changes
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 groups assigned to application: {appId}")
from fctr_okta_mcp.client.base_okta_api_client import OktaAPIClient
client = OktaAPIClient(ctx=ctx)
params = {"limit": 3} # Hardcoded for sample data
if q:
params["q"] = q
if after:
params["after"] = after
if limit and limit > 0:
params["limit"] = min(limit, 3)
if expand:
params["expand"] = expand
await ctx.info(f"Calling Okta API: GET /api/v1/apps/{appId}/groups...")
result = await client.make_request(
method="GET",
endpoint=f"/api/v1/apps/{appId}/groups",
params=params,
max_results=3
)
if result.get("status") == "success":
data = result.get("data", [])
await ctx.info(f"Retrieved {len(data)} group assignment(s)")
result["_reminder"] = "This is SAMPLE DATA (3 results). Generate Python code using execute_code() to get full results."
log_audit_event(
action="okta_app_list_groups",
status="success",
details={"app_id": appId, "result_count": len(data)}
)
else:
await ctx.warning(f"API returned status: {result.get('status')}")
log_audit_event(
action="okta_app_list_groups",
status="error",
details={"app_id": appId, "api_status": result.get("status")}
)
return compress_response(result)
logger.info("Application tools registered successfully")
