from typing import List, Optional
# Example 1: Information Retrieval Tool
# This tool retrieves data scoped to the user without modifying anything
@mcp.tool()
async def get_user_enrolled_courses(
include_completed: Optional[bool] = True,
category: Optional[str] = None
) -> dict:
"""
Get courses the current user is enrolled in.
Returns list of active and optionally completed courses.
"""
# Extract current user from headers (not a parameter!)
headers = get_http_headers()
user_id = headers.get('x-user-id') # This is the user making the request
try:
# Fetch courses for this user (internal implementation)
courses = fetch_user_courses(user_id, include_completed, category) # Your implementation
if not courses:
return {
"status": "success",
"message": "No courses found",
"data": {
"courses": [],
"total": 0
}
}
# Build response without exposing internal user IDs or database keys
course_list = []
for course in courses:
course_list.append({
"course_id": course["public_id"], # Use public identifiers
"title": course["title"],
"progress": course["progress_percentage"],
"status": course["status"],
"last_accessed": course["last_accessed"]
})
return {
"status": "success",
"message": f"Found {len(course_list)} enrolled courses",
"data": {
"courses": course_list,
"total": len(course_list),
"filters_applied": {
"include_completed": include_completed,
"category": category
}
}
}
except ValueError as e:
# User-friendly error for invalid inputs
return {
"status": "error",
"message": f"Invalid filter parameters: {str(e)}",
"data": None
}
except PermissionError:
# Clear message about access restrictions
return {
"status": "error",
"message": "Unable to access course enrollment data",
"data": None
}
except Exception as e:
# Generic error - log internally but don't expose details
print(f"Course fetch error for user {user_id}: {e}") # Internal logging only
return {
"status": "error",
"message": "Unable to retrieve courses. Please try again.",
"data": None
}
# Example 2: Admin Tool - Managing Other Users
# This tool allows an admin to create and manage OTHER users
@mcp.tool()
async def create_user(
user_name: str,
email: str
) -> dict:
"""
Admin tool: Create a new user account.
Requires admin permissions.
"""
headers = get_http_headers()
admin_id = headers.get('x-user-id') # The ADMIN performing this action
# Verify admin permissions
if not is_user_admin(admin_id):
return {
"status": "error",
"message": "Admin permissions required"
}
# Create the new user (email is the TARGET user, not the admin)
result = create_new_user(user_name, email)
return {
"status": "success",
"message": f"User '{user_name}' created successfully",
"data": {
"user_id": result["id"], # Return ID for subsequent operations
"name": user_name,
"email": email
}
}
# Example 3: Admin Tool - Modifying Another User's Settings
@mcp.tool()
async def update_user_settings(
user_id: str, # TARGET user whose settings to update
setting_name: str,
value: str
) -> dict:
"""
Change settings (theme, language, notifications) for a specific user.
Use when modifying settings for a user other than the current user.
"""
headers = get_http_headers()
admin_id = headers.get('x-user-id') # The ADMIN performing this action
# Verify admin permissions
if not is_user_admin(admin_id):
return {
"status": "error",
"message": "Admin permissions required"
}
valid_settings = ["theme", "language", "notifications"]
if setting_name not in valid_settings:
return {
"status": "error",
"message": f"Unknown setting '{setting_name}'. Valid options: {', '.join(valid_settings)}"
}
# Update the TARGET user's settings (not the admin's)
# Note: The user_id parameter identifies a different user than the one in x-user-id header
# This user_id would have been identified through prior conversation or tool calls
update_user_setting(user_id, setting_name, value) # Your implementation
return {
"status": "success",
"message": f"Settings updated for user {user_id}"
}
