get_peer_review_assignments
Retrieve peer review assignments and completion status for Canvas courses to track reviewer-reviewee relationships and submission progress.
Instructions
Get comprehensive peer review assignment mapping showing who is assigned to review whom with accurate completion status.
Args:
course_identifier: Canvas course code (e.g., badm_554_120251_246794) or ID
assignment_id: Canvas assignment ID
include_names: Include student names (requires additional API call)
include_submission_details: Include submission metadata
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| course_identifier | Yes | ||
| assignment_id | Yes | ||
| include_names | No | ||
| include_submission_details | No |
Implementation Reference
- The main execution handler for the 'get_peer_review_assignments' MCP tool. Decorated with @mcp.tool() for registration and @validate_params for input validation. Handles course resolution, analyzer invocation, error handling, and JSON response formatting.@mcp.tool() @validate_params async def get_peer_review_assignments( course_identifier: str | int, assignment_id: str | int, include_names: bool = True, include_submission_details: bool = False ) -> str: """Get comprehensive peer review assignment mapping showing who is assigned to review whom with accurate completion status. Args: course_identifier: Canvas course code (e.g., badm_554_120251_246794) or ID assignment_id: Canvas assignment ID include_names: Include student names (requires additional API call) include_submission_details: Include submission metadata """ try: course_id = await get_course_id(course_identifier) analyzer = PeerReviewAnalyzer() result = await analyzer.get_assignments( course_id=course_id, assignment_id=int(assignment_id), include_names=include_names, include_submission_details=include_submission_details ) if "error" in result: return f"Error getting peer review assignments: {result['error']}" return json.dumps(result, indent=2) except Exception as e: return f"Error in get_peer_review_assignments: {str(e)}"
- src/canvas_mcp/server.py:52-52 (registration)Call to register_peer_review_tools(mcp) in the main server registration function, which registers all peer review tools including 'get_peer_review_assignments'.register_peer_review_tools(mcp)
- Core helper method PeerReviewAnalyzer.get_assignments() that performs Canvas API calls to retrieve assignment details, peer reviews, and user data, then processes and structures the peer review assignments mapping with status information.async def get_assignments( self, course_id: int, assignment_id: int, include_names: bool = True, include_submission_details: bool = False ) -> dict[str, Any]: """Get peer review assignments with clear reviewer-reviewee mapping.""" 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 peer_reviews_response = await make_canvas_request( "get", f"/courses/{course_id}/assignments/{assignment_id}/peer_reviews" ) 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 if names are requested users_map = {} if include_names: 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.get("name", "Unknown") for user in users_response} # Process peer review data assignments_list = [] 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 assignment_entry = { "reviewer_id": reviewer_id, "reviewee_id": reviewee_id, "submission_id": pr.get("submission_id"), "status": "completed" if pr.get("workflow_state") == "completed" else "assigned", "assigned_date": pr.get("created_at"), "completed_date": pr.get("updated_at") if pr.get("workflow_state") == "completed" else None, "has_comments": bool(pr.get("comment")), "rubric_completed": bool(pr.get("rubric_assessment_id")), "peer_review_id": pr.get("id") } if include_names: assignment_entry["reviewer_name"] = users_map.get(reviewer_id, "Unknown") assignment_entry["reviewee_name"] = users_map.get(reviewee_id, "Unknown") assignments_list.append(assignment_entry) # Calculate peer review settings peer_review_settings = { "anonymous": assignment_response.get("anonymous_peer_reviews", False), "automatic": assignment_response.get("automatic_peer_reviews", False), "reviews_per_student": assignment_response.get("peer_review_count", 0) } return { "assignment_info": { "id": assignment_response.get("id"), "name": assignment_response.get("name"), "course_id": course_id, "total_reviews_assigned": len(assignments_list), "peer_review_settings": peer_review_settings }, "assignments": assignments_list } except Exception as e: return {"error": f"Exception in get_assignments: {str(e)}"}