Canvas MCP Server

Instructions

    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
    

Implementation Reference

• src/canvas_mcp/tools/peer_reviews.py:15-48 (handler)

  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)

• src/canvas_mcp/core/peer_reviews.py:20-106 (helper)

  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)}"}