"""Student-specific MCP tools for Canvas API.
These tools provide student-focused functionality using Canvas API "/self" endpoints
to access only the student's own data across their enrolled courses.
"""
from datetime import datetime, timedelta, timezone
from typing import Any
from mcp.server.fastmcp import FastMCP
from ..core.cache import get_course_code, get_course_id
from ..core.client import fetch_all_paginated_results, make_canvas_request
from ..core.dates import format_date, parse_date
from ..core.validation import validate_params
def register_student_tools(mcp: FastMCP):
"""Register student-specific MCP tools."""
@mcp.tool()
async def get_my_upcoming_assignments(days: int = 7) -> str:
"""Get your upcoming assignments across all courses.
Args:
days: Number of days to look ahead (default: 7)
Returns upcoming assignments due within the specified timeframe,
sorted by due date, with submission status.
"""
# Calculate the date range (use timezone-aware datetime)
end_date = datetime.now(timezone.utc) + timedelta(days=days)
end_date_str = end_date.strftime("%Y-%m-%d")
# Get upcoming events for the current user
events = await fetch_all_paginated_results(
"/users/self/upcoming_events",
params={"per_page": 100}
)
if isinstance(events, dict) and "error" in events:
return f"Error fetching upcoming assignments: {events['error']}"
if not events:
return f"No assignments due in the next {days} days."
# Filter to assignments only (not calendar events)
assignments = []
for event in events:
if event.get("type") == "assignment" or event.get("assignment"):
assignment_data = event.get("assignment", event)
due_at = assignment_data.get("due_at")
if due_at:
# Check if within our date range
due_date = parse_date(due_at)
if due_date and due_date <= end_date:
assignments.append(assignment_data)
if not assignments:
return f"No assignments due in the next {days} days."
# Sort by due date (use timezone-aware max for fallback)
assignments.sort(key=lambda x: parse_date(x.get("due_at", "")) or datetime.max.replace(tzinfo=timezone.utc))
# Format output
output_lines = [f"Upcoming Assignments (Next {days} Days):\n"]
for assignment in assignments:
name = assignment.get("name", "Unnamed Assignment")
due_at = format_date(assignment.get("due_at"))
course_id = assignment.get("course_id")
# Get course name
course_display = await get_course_code(course_id) if course_id else "Unknown Course"
# Get submission status
submission = assignment.get("submission")
if submission:
submitted = submission.get("submitted_at") is not None
status = "✅ Submitted" if submitted else "❌ Not Submitted"
else:
status = "❌ Not Submitted"
output_lines.append(
f"• {name}\n"
f" Course: {course_display}\n"
f" Due: {due_at}\n"
f" Status: {status}\n"
)
return "\n".join(output_lines)
@mcp.tool()
@validate_params
async def get_my_submission_status(course_identifier: str | int | None = None) -> str:
"""Get your submission status for assignments.
Args:
course_identifier: Optional course code or ID to filter by specific course.
If not provided, shows all courses.
Returns your submission status across assignments, highlighting missing submissions.
"""
if course_identifier:
# Get submissions for specific course
course_id = await get_course_id(course_identifier)
assignments = await fetch_all_paginated_results(
f"/courses/{course_id}/assignments",
params={"include[]": ["submission"], "per_page": 100}
)
if isinstance(assignments, dict) and "error" in assignments:
return f"Error fetching assignments: {assignments['error']}"
course_display = await get_course_code(course_id) or course_identifier
output_lines = [f"Submission Status for {course_display}:\n"]
else:
# Get all courses and their assignments
courses = await fetch_all_paginated_results(
"/courses",
params={"enrollment_state": "active", "per_page": 100}
)
if isinstance(courses, dict) and "error" in courses:
return f"Error fetching courses: {courses['error']}"
output_lines = ["Submission Status (All Courses):\n"]
all_assignments = []
for course in courses:
course_id = course.get("id")
course_name = course.get("course_code", course.get("name", "Unknown"))
assignments = await fetch_all_paginated_results(
f"/courses/{course_id}/assignments",
params={"include[]": ["submission"], "per_page": 100}
)
if not isinstance(assignments, dict) or "error" not in assignments:
for assignment in assignments if isinstance(assignments, list) else []:
assignment["_course_name"] = course_name
all_assignments.append(assignment)
assignments = all_assignments
if not assignments:
return "No assignments found."
# Separate submitted and missing
submitted = []
missing = []
for assignment in assignments:
submission = assignment.get("submission")
is_submitted = submission and submission.get("submitted_at") is not None
if is_submitted:
submitted.append(assignment)
else:
# Check if past due (use timezone-aware datetime)
due_at = assignment.get("due_at")
if due_at:
due_date = parse_date(due_at)
if due_date and due_date < datetime.now(timezone.utc):
missing.append((assignment, "OVERDUE"))
else:
missing.append((assignment, "NOT SUBMITTED"))
else:
missing.append((assignment, "NOT SUBMITTED"))
# Format output
if missing:
output_lines.append(f"⚠️ Missing Submissions ({len(missing)}):\n")
for assignment, status in missing:
name = assignment.get("name", "Unnamed")
due_at = format_date(assignment.get("due_at")) if assignment.get("due_at") else "No due date"
course_name = assignment.get("_course_name", "")
output_lines.append(
f"• {name}\n"
f" {f'Course: {course_name}' if course_name else ''}\n"
f" Due: {due_at}\n"
f" Status: {status}\n"
)
if submitted:
output_lines.append(f"\n✅ Submitted ({len(submitted)}):\n")
for assignment in submitted[:10]: # Show first 10
name = assignment.get("name", "Unnamed")
submission = assignment.get("submission", {})
submitted_at = format_date(submission.get("submitted_at"))
course_name = assignment.get("_course_name", "")
output_lines.append(
f"• {name}\n"
f" {f'Course: {course_name}' if course_name else ''}\n"
f" Submitted: {submitted_at}\n"
)
return "\n".join(output_lines)
@mcp.tool()
async def get_my_course_grades() -> str:
"""Get your current grades across all enrolled courses.
Returns your current grade, enrollment status, and recent performance
for each active course.
"""
courses = await fetch_all_paginated_results(
"/courses",
params={
"enrollment_state": "active",
"include[]": ["total_scores", "current_grading_period_scores"],
"per_page": 100
}
)
if isinstance(courses, dict) and "error" in courses:
return f"Error fetching courses: {courses['error']}"
if not courses:
return "No active course enrollments found."
output_lines = ["Your Course Grades:\n"]
for course in courses:
name = course.get("name", "Unnamed Course")
course_code = course.get("course_code", "")
# Get enrollment data (grades)
enrollments = course.get("enrollments", [])
if enrollments:
enrollment = enrollments[0] # Student typically has one enrollment per course
# Current score
current_score = enrollment.get("computed_current_score")
final_score = enrollment.get("computed_final_score")
current_grade = enrollment.get("computed_current_grade", "N/A")
# Format grade info
if current_score is not None:
grade_info = f"{current_grade} ({current_score:.1f}%)"
elif final_score is not None:
grade_info = f"{final_score:.1f}%"
else:
grade_info = "No grade yet"
output_lines.append(
f"• {course_code}: {name}\n"
f" Current Grade: {grade_info}\n"
)
else:
output_lines.append(
f"• {course_code}: {name}\n"
f" Current Grade: No enrollment data\n"
)
return "\n".join(output_lines)
@mcp.tool()
async def get_my_todo_items() -> str:
"""Get your Canvas TODO list.
Returns all items in your Canvas TODO list including assignments,
quizzes, and discussions that need your attention.
"""
todos = await fetch_all_paginated_results(
"/users/self/todo",
params={"per_page": 100}
)
if isinstance(todos, dict) and "error" in todos:
return f"Error fetching TODO items: {todos['error']}"
if not todos:
return "Your TODO list is empty! 🎉"
output_lines = ["Your TODO List:\n"]
for item in todos:
item_type = item.get("type", "item")
assignment = item.get("assignment", {})
name = assignment.get("name") or item.get("title", "Unnamed item")
due_at = format_date(assignment.get("due_at")) if assignment.get("due_at") else "No due date"
course_id = item.get("course_id")
course_display = await get_course_code(course_id) if course_id else "Unknown Course"
output_lines.append(
f"• {name}\n"
f" Type: {item_type.title()}\n"
f" Course: {course_display}\n"
f" Due: {due_at}\n"
)
return "\n".join(output_lines)
@mcp.tool()
@validate_params
async def get_my_peer_reviews_todo(course_identifier: str | int | None = None) -> str:
"""Get peer reviews you need to complete.
Args:
course_identifier: Optional course code or ID to filter by specific course
Returns list of peer reviews assigned to you that need completion.
"""
if course_identifier:
course_ids = [await get_course_id(course_identifier)]
else:
# Get all active courses
courses = await fetch_all_paginated_results(
"/courses",
params={"enrollment_state": "active", "per_page": 100}
)
if isinstance(courses, dict) and "error" in courses:
return f"Error fetching courses: {courses['error']}"
course_ids = [course.get("id") for course in courses if course.get("id")]
all_peer_reviews = []
for course_id in course_ids:
# Get assignments for this course
assignments = await fetch_all_paginated_results(
f"/courses/{course_id}/assignments",
params={"per_page": 100}
)
if isinstance(assignments, dict) and "error" in assignments:
continue
# Check each assignment for peer reviews
for assignment in assignments if isinstance(assignments, list) else []:
if assignment.get("peer_reviews"):
assignment_id = assignment.get("id")
# Get peer reviews for this assignment
peer_reviews = await fetch_all_paginated_results(
f"/courses/{course_id}/assignments/{assignment_id}/peer_reviews",
params={"include[]": ["user"], "per_page": 100}
)
if isinstance(peer_reviews, list):
# Filter to reviews assigned to current user that are incomplete
for review in peer_reviews:
# Note: We'd need to filter by current user ID
# For now, show all incomplete reviews
if review.get("workflow_state") != "completed":
review["_course_id"] = course_id
review["_assignment_name"] = assignment.get("name")
all_peer_reviews.append(review)
if not all_peer_reviews:
return "You have no pending peer reviews! ✅"
output_lines = ["Peer Reviews You Need to Complete:\n"]
for review in all_peer_reviews:
assignment_name = review.get("_assignment_name", "Unknown Assignment")
course_id = review.get("_course_id")
course_display = await get_course_code(course_id) if course_id else "Unknown Course"
user_id = review.get("user_id")
assessor_id = review.get("assessor_id")
output_lines.append(
f"• {assignment_name}\n"
f" Course: {course_display}\n"
f" Reviewing: Student {user_id}\n"
f" Status: Incomplete\n"
)
return "\n".join(output_lines)
