# src/fctr_okta_mcp/tools/okta/group_tools.py
"""
Group Management Tools for Okta MCP Server.
Tools for listing, searching, and managing Okta groups,
including group members and application 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_group_tools(mcp: FastMCP):
"""Register all group-related tools with the MCP server."""
logger.debug("Registering group tools...")
@mcp.tool(
tags={"okta", "group", "direct"},
annotations={
"category": "group",
"readOnlyHint": True,
"idempotentHint": True
}
)
async def okta_group_list(
ctx: Context,
search: Annotated[str, "SCIM expressions for optimal search. Searches profile attributes (prefix with 'profile.'), system properties (id, type, created, lastUpdated, lastMembershipUpdated), and source.id for APP_GROUP types. Example: 'profile.name eq \"Engineering Team\" and type eq \"OKTA_GROUP\"'. Supports: sw, eq, co (limited to profile.name and profile.description only). Use 'and', 'or' for combining."] = "",
filter: Annotated[str, "SCIM expressions for limited system properties ONLY: id, type, lastUpdated, lastMembershipUpdated. Example: 'type eq \"OKTA_GROUP\" and lastUpdated gt \"2024-01-01T00:00:00.000Z\"'. Cannot combine with 'search'."] = "",
q: Annotated[str, "Simple search matching group name property only. Example: 'Sales'. Note: 300 result limit, no pagination support. Cannot combine with 'search'."] = "",
after: Annotated[str, "Pagination cursor from previous response's _links.next for fetching next page. Not available with search queries."] = "",
limit: Annotated[int, "Number of results per page (max: 10000, recommended: ≤200). Default varies by query type."] = 0,
expand: Annotated[str, "Include metadata in _embedded property. Values: 'stats' (user count as usersCount) or 'app' (application details for APP_GROUP types)."] = "",
sortBy: Annotated[str, "Sort field for search queries only. Any single property like 'profile.name', 'lastUpdated', 'created'."] = "",
sortOrder: Annotated[str, "Sort direction 'asc' or 'desc' (search queries only, requires sortBy)."] = ""
) -> 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 groups in the organization. With expand=stats, provides member counts in single call (replaces list-group-members count queries). With expand=app, includes assigned application details for APP_GROUP types (replaces list-assigned-apps-for-group calls). Use for group management, membership analysis, access governance, and organizational structure reviews.
## Query Parameters
- **search**: Most powerful method using SCIM expressions for optimal search performance. Searches profile attributes (prefix with `profile.`), system properties (id, type, created, lastUpdated, lastMembershipUpdated), and source.id for APP_GROUP types.
- **filter**: SCIM expressions for limited system properties only: id, type, lastUpdated, lastMembershipUpdated.
- **q**: Simple search matching group name property only. Default limit 300 results, pagination not supported.
- **expand**: Include metadata - `stats` (user count) or `app` (application details for APP_GROUP).
- **limit**: Number of results per page (max: 10000, recommended: ≤200).
- **after**: Pagination cursor for next page (not available with search queries).
- **sortBy**: Sort field for search queries only.
- **sortOrder**: Sort direction `asc` or `desc` (search queries only, requires sortBy).
## Parameter Usage Notes
- **CRITICAL**: Use `search` for profile attributes, use `filter` for limited system properties only - mixing causes errors.
- Search and pagination are mutually exclusive - search queries cannot use `after` parameter.
- Query parameter (`q`) has 300 result limit, no pagination support.
- Use `expand=stats` to get member counts without additional API calls.
- Use `expand=app` for APP_GROUP types to get application assignments in single call.
## Search Examples
- Group Type: `search=type eq "OKTA_GROUP"` or `search=type eq "APP_GROUP"`
- Profile Attributes: `search=profile.name eq "Engineering Team"`
- Time-based: `search=lastMembershipUpdated gt "2024-01-01T00:00:00.000Z"`
- Source Application: `search=source.id eq "0oa2v0el0gP90aqjJ0g7"` (APP_GROUP only)
- Complex: `search=type eq "APP_GROUP" and created lt "2024-01-01T00:00:00.000Z"`
- Name Contains: `search=profile.name co "Sales"` (partial matching)
- Name Starts With: `search=profile.samAccountName sw "West Coast"`
## Filter Examples
- System Properties: `filter=type eq "OKTA_GROUP"`
- Multiple Criteria: `filter=type eq "OKTA_GROUP" and lastUpdated gt "2024-01-01T00:00:00.000Z"`
## Important Notes
- The `co` operator only works with profile.name and profile.description - other attributes must use `eq` or `sw`.
- All SCIM expressions must be URL encoded.
- Search uses eventually consistent datasource - may not reflect latest changes immediately.
- Default Everyone group not returned for users with group admin role.
"""
# 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 groups ({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 limit and limit > 0:
params["limit"] = min(limit, 3) # Cap at 3 for sample
if expand:
params["expand"] = expand
if sortBy:
params["sortBy"] = sortBy
if sortOrder:
params["sortOrder"] = sortOrder
await ctx.info("Calling Okta API: GET /api/v1/groups...")
result = await client.make_request(
method="GET",
endpoint="/api/v1/groups",
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 group(s)")
result["_reminder"] = "This is SAMPLE DATA (3 results). Generate Python code using execute_code() to get full results."
log_audit_event(
action="okta_group_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_group_list",
status="error",
details={"filter": filter_log, "api_status": result.get("status")}
)
return compress_response(result)
@mcp.tool(
tags={"okta", "group", "direct"},
annotations={
"category": "group",
"readOnlyHint": True,
"idempotentHint": True
}
)
async def okta_group_get(
ctx: Context,
groupId: Annotated[str, "The ID of the group to retrieve (required)"]
) -> dict:
"""
**CALL read_system_instructions() FIRST!**
Get detailed information about a specific group by its ID.
## Path Parameters
- **groupId** (String, Required): The unique ID of the group.
## Response Fields
- **id**: Group's unique identifier
- **created**: When the group was created
- **lastUpdated**: When the group was last modified
- **lastMembershipUpdated**: When membership was last changed
- **type**: Group type (OKTA_GROUP, APP_GROUP, BUILT_IN)
- **profile.name**: Group display name
- **profile.description**: Group description
- **_embedded**: Additional data if expand parameter used
## Group Types
- **OKTA_GROUP**: Groups managed within Okta
- **APP_GROUP**: Groups synced from applications (AD, LDAP, etc.)
- **BUILT_IN**: System groups like "Everyone"
## Common Use Cases
- Get group details before modifications
- Verify group exists
- Check group type and metadata
- Audit group information
"""
await ctx.info(f"Fetching group details: {groupId}")
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/groups/{groupId}...")
result = await client.make_request(
method="GET",
endpoint=f"/api/v1/groups/{groupId}",
params={}
)
if result.get("status") == "success":
data = result.get("data", {})
group_name = data.get("profile", {}).get("name", "unknown")
group_type = data.get("type", "unknown")
await ctx.info(f"Retrieved group: {group_name} (type: {group_type})")
log_audit_event(
action="okta_group_get",
status="success",
details={"group_id": groupId, "name": group_name, "type": group_type}
)
else:
await ctx.warning(f"API returned status: {result.get('status')}")
log_audit_event(
action="okta_group_get",
status="error",
details={"group_id": groupId, "api_status": result.get("status")}
)
return compress_response(result)
@mcp.tool(
tags={"okta", "group", "user", "direct"},
annotations={
"category": "group",
"readOnlyHint": True,
"idempotentHint": True
}
)
async def okta_group_list_users(
ctx: Context,
groupId: Annotated[str, "The ID of the group (required)"],
after: Annotated[str, "Pagination cursor from previous response for fetching next page."] = ""
) -> dict:
"""
**CALL read_system_instructions() FIRST!**
Lists all users who are members of the specified group.
## Path Parameters
- **groupId** (String, Required): The ID of the group.
## Query Parameters
- **after** (String, Optional): Pagination cursor from previous response.
- **limit** (Integer, Optional): Number of results per page. (Hardcoded to 3 for sample)
## Response Fields
Returns array of user objects, each containing:
- **id**: User's unique identifier
- **status**: User status (ACTIVE, STAGED, etc.)
- **created**: When user was created
- **profile**: User profile (firstName, lastName, email, login)
- **credentials**: Authentication credentials info
## Common Use Cases
- Get all members of a group
- Audit group membership
- Export group member list
- Compare memberships across groups
- Access review for group-based permissions
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 for group: {groupId}")
from fctr_okta_mcp.client.base_okta_api_client import OktaAPIClient
client = OktaAPIClient(ctx=ctx)
params = {"limit": 3} # Hardcoded for sample data
if after:
params["after"] = after
await ctx.info(f"Calling Okta API: GET /api/v1/groups/{groupId}/users...")
result = await client.make_request(
method="GET",
endpoint=f"/api/v1/groups/{groupId}/users",
params=params,
max_results=3
)
if result.get("status") == "success":
data = result.get("data", [])
user_emails = [u.get("profile", {}).get("email", "unknown") for u in data]
await ctx.info(f"Retrieved {len(data)} user(s): {', '.join(user_emails) if user_emails 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_group_list_users",
status="success",
details={"group_id": groupId, "result_count": len(data)}
)
else:
await ctx.warning(f"API returned status: {result.get('status')}")
log_audit_event(
action="okta_group_list_users",
status="error",
details={"group_id": groupId, "api_status": result.get("status")}
)
return compress_response(result)
@mcp.tool(
tags={"okta", "group", "application", "direct"},
annotations={
"category": "group",
"readOnlyHint": True,
"idempotentHint": True
}
)
async def okta_group_list_applications(
ctx: Context,
groupId: Annotated[str, "The ID of the group (required)"],
after: Annotated[str, "Pagination cursor from previous response for fetching next page."] = ""
) -> dict:
"""
**CALL read_system_instructions() FIRST!**
Lists all applications assigned to the specified group.
## Path Parameters
- **groupId** (String, Required): The ID of the group.
## Query Parameters
- **after** (String, Optional): Pagination cursor from previous response.
## Response Fields
Returns array of application objects, each containing:
- **id**: Application's unique identifier
- **name**: Application internal name
- **label**: Application display name
- **status**: Application status (ACTIVE, INACTIVE)
- **signOnMode**: How users authenticate (SAML, OIDC, etc.)
- **features**: Enabled features list
- **created**: When application was created
- **lastUpdated**: When application was last modified
## Common Use Cases
- Audit which apps a group provides access to
- Review group-based application assignments
- Access governance and compliance reporting
- 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 applications assigned to group: {groupId}")
from fctr_okta_mcp.client.base_okta_api_client import OktaAPIClient
client = OktaAPIClient(ctx=ctx)
params = {"limit": 3} # Hardcoded for sample data
if after:
params["after"] = after
await ctx.info(f"Calling Okta API: GET /api/v1/groups/{groupId}/apps...")
result = await client.make_request(
method="GET",
endpoint=f"/api/v1/groups/{groupId}/apps",
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)} application(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_group_list_applications",
status="success",
details={"group_id": groupId, "result_count": len(data)}
)
else:
await ctx.warning(f"API returned status: {result.get('status')}")
log_audit_event(
action="okta_group_list_applications",
status="error",
details={"group_id": groupId, "api_status": result.get("status")}
)
return compress_response(result)
logger.info("Group tools registered successfully")
