Canvas MCP Server

Instructions

    Args:
        course_identifier: The Canvas course code (e.g., badm_554_120251_246794) or ID
        topic_id: The Canvas discussion topic ID
        include_replies: Whether to fetch detailed replies for all entries (default: False)
    

Implementation Reference

• src/canvas_mcp/tools/discussions.py:535-662 (handler)

  The core handler function for the 'get_discussion_with_replies' MCP tool. Fetches discussion topic entries using Canvas API, optionally retrieves replies using multiple endpoints (entries, view, replies), anonymizes data, formats output with previews or full content.

  @mcp.tool()
  @validate_params
  async def get_discussion_with_replies(course_identifier: str | int,
                                      topic_id: str | int,
                                      include_replies: bool = False) -> str:
      """Enhanced function to get discussion entries with optional reply fetching.
  
      Args:
          course_identifier: The Canvas course code (e.g., badm_554_120251_246794) or ID
          topic_id: The Canvas discussion topic ID
          include_replies: Whether to fetch detailed replies for all entries (default: False)
      """
      course_id = await get_course_id(course_identifier)
  
      # Get basic entries first
      entries = await fetch_all_paginated_results(
          f"/courses/{course_id}/discussion_topics/{topic_id}/entries",
          {"per_page": 100}
      )
  
      if isinstance(entries, dict) and "error" in entries:
          return f"Error fetching discussion entries: {entries['error']}"
  
      if not entries:
          return f"No discussion entries found for topic {topic_id}."
  
      # Get topic details for context
      topic_response = await make_canvas_request(
          "get", f"/courses/{course_id}/discussion_topics/{topic_id}"
      )
  
      topic_title = "Unknown Topic"
      if "error" not in topic_response:
          topic_title = topic_response.get("title", "Unknown Topic")
  
      course_display = await get_course_code(course_id) or course_identifier
      result = f"Discussion '{topic_title}' in Course {course_display}:\n\n"
  
      # Process each entry
      for entry in entries:
          entry_id = entry.get("id")
          user_name = entry.get("user_name", "Unknown user")
          message = entry.get("message", "")
          created_at = format_date(entry.get("created_at"))
  
          # Clean up message for display
          import re
          if message:
              message_preview = re.sub(r'<[^>]+>', '', message)
              if len(message_preview) > 200:
                  message_preview = message_preview[:200] + "..."
              message_preview = message_preview.replace("\n", " ").strip()
          else:
              message_preview = "[No content]"
  
          result += f"📝 Entry {entry_id} by {user_name}\n"
          result += f"   Posted: {created_at}\n"
          result += f"   Content: {message_preview}\n"
  
          # Handle replies
          if include_replies:
              replies = []
  
              # Method 1: Check recent_replies from the entry
              recent_replies = entry.get("recent_replies", [])
              if recent_replies:
                  replies = recent_replies
  
              # Method 2: If no recent_replies or has_more_replies, try direct API call
              has_more_replies = entry.get("has_more_replies", False)
              if not replies or has_more_replies:
                  try:
                      replies_response = await fetch_all_paginated_results(
                          f"/courses/{course_id}/discussion_topics/{topic_id}/entries/{entry_id}/replies",
                          {"per_page": 100}
                      )
  
                      if not isinstance(replies_response, dict) or "error" not in replies_response:
                          replies = replies_response
                  except Exception as e:
                      log_warning(
                          "Failed to fetch detailed replies",
                          exc=e,
                          course_id=course_id,
                          topic_id=topic_id,
                          entry_id=entry_id
                      )
  
              # Display replies
              if replies:
                  result += f"   💬 Replies ({len(replies)}):\n"
                  for i, reply in enumerate(replies, 1):
                      reply_user = reply.get("user_name", "Unknown")
                      reply_created = format_date(reply.get("created_at"))
                      reply_msg = reply.get("message", "")
  
                      # Clean reply message
                      if reply_msg:
                          reply_preview = re.sub(r'<[^>]+>', '', reply_msg)
                          if len(reply_preview) > 150:
                              reply_preview = reply_preview[:150] + "..."
                          reply_preview = reply_preview.replace("\n", " ").strip()
                      else:
                          reply_preview = "[No content]"
  
                      result += f"      └─ Reply {i} by {reply_user} ({reply_created}): {reply_preview}\n"
              else:
                  recent_count = len(entry.get("recent_replies", []))
                  has_more = entry.get("has_more_replies", False)
                  if recent_count > 0 or has_more:
                      result += f"   💬 Replies: {recent_count}{'+ (has more)' if has_more else ''} (failed to fetch details)\n"
                  else:
                      result += "   💬 No replies\n"
          else:
              # Just show reply count without fetching
              recent_count = len(entry.get("recent_replies", []))
              has_more = entry.get("has_more_replies", False)
              if recent_count > 0 or has_more:
                  result += f"   💬 Replies: {recent_count}{'+ (has more)' if has_more else ''}\n"
              else:
                  result += "   💬 No replies\n"
  
          result += "\n"
  
      if not include_replies:
          result += "\n💡 Tip: Use include_replies=True to fetch detailed reply content"
  
      return result

• src/canvas_mcp/server.py:49-49 (registration)

  Within register_all_tools, calls register_discussion_tools(mcp) which defines nested functions decorated with @mcp.tool(), thereby registering 'get_discussion_with_replies'.

  register_discussion_tools(mcp)

• src/canvas_mcp/tools/__init__.py:5-5 (registration)

  Imports register_discussion_tools from discussions.py for use in tool registration.

  from .discussions import register_discussion_tools