Canvas MCP Server

Instructions

Implementation Reference

• src/canvas_mcp/tools/peer_review_comments.py:20-60 (handler)

  The main MCP tool handler for 'get_peer_review_comments', registered via @mcp.tool() decorator. Handles input parameters, calls the core analyzer, and formats output as JSON.
  @mcp.tool() @validate_params async def get_peer_review_comments( course_identifier: str | int, assignment_id: str | int, include_reviewer_info: bool = True, include_reviewee_info: bool = True, include_submission_context: bool = False, anonymize_students: bool = False ) -> str: """ Retrieve actual comment text for peer reviews on a specific assignment. Args: course_identifier: Canvas course code (e.g., badm_554_120251_246794) or ID assignment_id: Canvas assignment ID include_reviewer_info: Include reviewer student information include_reviewee_info: Include reviewee student information include_submission_context: Include original submission details anonymize_students: Replace student names with anonymous IDs """ try: course_id = await get_course_id(course_identifier) analyzer = PeerReviewCommentAnalyzer() result = await analyzer.get_peer_review_comments( course_id=course_id, assignment_id=int(assignment_id), include_reviewer_info=include_reviewer_info, include_reviewee_info=include_reviewee_info, include_submission_context=include_submission_context, anonymize_students=anonymize_students ) if "error" in result: return f"Error getting peer review comments: {result['error']}" return json.dumps(result, indent=2) except Exception as e: return f"Error in get_peer_review_comments: {str(e)}"
• src/canvas_mcp/core/peer_review_comments.py:40-242 (helper)

  Core implementation logic in PeerReviewCommentAnalyzer class that fetches peer review comments from Canvas API, processes them, handles anonymization, and computes statistics. Called by the tool handler.
  async def get_peer_review_comments( self, course_id: int, assignment_id: int, include_reviewer_info: bool = True, include_reviewee_info: bool = True, include_submission_context: bool = False, anonymize_students: bool = False ) -> dict[str, Any]: """ Retrieve actual comment text for peer reviews on a specific assignment. Args: course_id: Canvas course ID assignment_id: Canvas assignment ID include_reviewer_info: Include reviewer student information include_reviewee_info: Include reviewee student information include_submission_context: Include original submission details anonymize_students: Replace student names with anonymous IDs Returns: Dict containing assignment info, peer reviews with comments, and summary statistics """ try: # Get assignment details assignment_response = await make_canvas_request( "get", f"/courses/{course_id}/assignments/{assignment_id}" ) if "error" in assignment_response: return {"error": f"Failed to get assignment: {assignment_response['error']}"} # Get peer reviews (these don't include comments directly) peer_reviews_response = await make_canvas_request( "get", f"/courses/{course_id}/assignments/{assignment_id}/peer_reviews", params={"include[]": ["user", "assessor"]} ) if "error" in peer_reviews_response: return {"error": f"Failed to get peer reviews: {peer_reviews_response['error']}"} peer_reviews: list[Any] = peer_reviews_response if isinstance(peer_reviews_response, list) else [] # Get users for name mapping if needed users_map = {} if include_reviewer_info or include_reviewee_info: users_response = await fetch_all_paginated_results( f"/courses/{course_id}/users", {"enrollment_type[]": "student", "per_page": 100} ) if isinstance(users_response, list): users_map = {user["id"]: user for user in users_response} # Get ALL submissions with comments (needed to extract peer review comments) submissions_map = {} submissions_by_id = {} submissions_response = await fetch_all_paginated_results( f"/courses/{course_id}/assignments/{assignment_id}/submissions", {"include[]": ["submission_comments"], "per_page": 100} ) if isinstance(submissions_response, list): for sub in submissions_response: submissions_map[sub["user_id"]] = sub submissions_by_id[sub["id"]] = sub # Process peer review comments processed_reviews = [] total_comments = 0 comments_with_text = 0 empty_comments = 0 total_word_count = 0 for pr in peer_reviews: reviewer_id = pr.get("assessor_id") reviewee_id = pr.get("user_id") if not reviewer_id or not reviewee_id: continue # Build reviewer info reviewer_info = {"student_id": reviewer_id} if include_reviewer_info and reviewer_id in users_map: user = users_map[reviewer_id] if anonymize_students: reviewer_info.update({ "student_name": generate_anonymous_id(reviewer_id, "Student"), "anonymous_id": generate_anonymous_id(reviewer_id, "Reviewer") }) else: reviewer_info.update({ "student_name": user.get("name", "Unknown"), "anonymous_id": generate_anonymous_id(reviewer_id, "Reviewer") }) # Build reviewee info reviewee_info = {"student_id": reviewee_id} if include_reviewee_info and reviewee_id in users_map: user = users_map[reviewee_id] if anonymize_students: reviewee_info.update({ "student_name": generate_anonymous_id(reviewee_id, "Student"), "anonymous_id": generate_anonymous_id(reviewee_id, "Reviewee") }) else: reviewee_info.update({ "student_name": user.get("name", "Unknown"), "anonymous_id": generate_anonymous_id(reviewee_id, "Reviewee") }) # Build submission info submission_info = {} if include_submission_context and reviewee_id in submissions_map: sub = submissions_map[reviewee_id] submission_info = { "submission_id": sub.get("id"), "submitted_at": format_date(sub.get("submitted_at")), "attempt": sub.get("attempt", 1) } # Process comment content - Extract from submission comments review_content: dict[str, Any] = { "comment_text": "", "rating": None, "rubric_assessments": [], "timestamp": None, "word_count": 0, "character_count": 0 } # Get the submission that was reviewed asset_id = pr.get("asset_id") # This is the submission ID if asset_id and asset_id in submissions_by_id: submission = submissions_by_id[asset_id] submission_comments = submission.get("submission_comments", []) # Find comments from this specific reviewer for comment in submission_comments: if comment.get("author_id") == reviewer_id: comment_text = comment.get("comment", "") review_content.update({ "comment_text": comment_text, "timestamp": format_date(comment.get("created_at")), "word_count": len(comment_text.split()) if comment_text else 0, "character_count": len(comment_text) if comment_text else 0 }) total_word_count += review_content["word_count"] if comment_text.strip(): comments_with_text += 1 else: empty_comments += 1 break # Use the first comment from this reviewer else: # No comment found from this reviewer empty_comments += 1 else: # No submission found for this asset_id empty_comments += 1 # Try to extract rating from rubric assessments if available # Note: This would require additional API calls to get rubric assessments # For now, we'll leave this as placeholder total_comments += 1 review_data = { "review_id": f"review_{pr.get('id', 'unknown')}", "reviewer": reviewer_info, "reviewee": reviewee_info, "review_content": review_content } if include_submission_context: review_data["submission_info"] = submission_info processed_reviews.append(review_data) # Calculate summary statistics avg_word_count = total_word_count / total_comments if total_comments > 0 else 0 result = { "assignment_info": { "assignment_id": assignment_id, "assignment_name": assignment_response.get("name", "Unknown"), "total_reviews": len(peer_reviews), "completed_reviews": total_comments }, "peer_reviews": processed_reviews, "summary_statistics": { "total_comments": total_comments, "average_word_count": round(avg_word_count, 1), "comments_with_text": comments_with_text, "empty_comments": empty_comments, "average_rating": None # Placeholder for future rubric integration } } return result except Exception as e: return {"error": f"Failed to get peer review comments: {str(e)}"}
• src/canvas_mcp/server.py:53-53 (registration)

  Registration call in the main server setup that invokes the module registration function containing the @mcp.tool() decorator for this tool.
  register_peer_review_comment_tools(mcp)