#!/usr/bin/env python3
"""Canvas MCP Server - Provides Canvas LMS integration for Claude Desktop."""
import asyncio
import json
import os
import sys
from typing import Any, Sequence
from dotenv import load_dotenv
import mcp.types as types
from mcp.server import NotificationOptions, Server
from mcp.server.models import InitializationOptions
import mcp.server.stdio
from canvas_api import CanvasAPI
# Load environment variables
load_dotenv()
class CanvasMCPServer:
"""MCP Server for Canvas LMS integration."""
def __init__(self):
self.server = Server("canvas-mcp-server")
self.canvas_api = None
async def initialize_canvas_api(self):
"""Initialize the Canvas API client."""
try:
self.canvas_api = CanvasAPI()
# Test the connection
await self.canvas_api.get_user_profile()
print("✅ Canvas API connection successful", file=sys.stderr)
except Exception as e:
print(f"❌ Failed to connect to Canvas API: {e}", file=sys.stderr)
raise
def setup_handlers(self):
"""Set up MCP request handlers."""
@self.server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
"""List available Canvas tools."""
return [
types.Tool(
name="get_courses",
description="Get all active Canvas courses for the user",
inputSchema={
"type": "object",
"properties": {},
"required": [],
},
),
types.Tool(
name="get_upcoming_assignments",
description="Get upcoming assignments and deadlines",
inputSchema={
"type": "object",
"properties": {
"days_ahead": {
"type": "number",
"description": "Number of days to look ahead (default: 30)",
"default": 30,
}
},
"required": [],
},
),
types.Tool(
name="get_course_assignments",
description="Get all assignments for a specific course",
inputSchema={
"type": "object",
"properties": {
"course_id": {
"type": "string",
"description": "Canvas course ID",
}
},
"required": ["course_id"],
},
),
types.Tool(
name="get_todo_items",
description="Get items from Canvas To-Do list",
inputSchema={
"type": "object",
"properties": {},
"required": [],
},
),
types.Tool(
name="get_assignment_details",
description="Get detailed information about a specific assignment",
inputSchema={
"type": "object",
"properties": {
"course_id": {
"type": "string",
"description": "Canvas course ID",
},
"assignment_id": {
"type": "string",
"description": "Canvas assignment ID",
},
},
"required": ["course_id", "assignment_id"],
},
),
types.Tool(
name="get_grades",
description="Get grades for all courses or a specific course",
inputSchema={
"type": "object",
"properties": {
"course_id": {
"type": "string",
"description": "Canvas course ID (optional - if not provided, gets grades for all courses)",
},
},
"required": [],
},
),
types.Tool(
name="get_calendar_events",
description="Get calendar events within a date range",
inputSchema={
"type": "object",
"properties": {
"start_date": {
"type": "string",
"description": "Start date in ISO format (optional)",
},
"end_date": {
"type": "string",
"description": "End date in ISO format (optional)",
},
},
"required": [],
},
),
types.Tool(
name="get_announcements",
description="Get announcements for a course or all courses",
inputSchema={
"type": "object",
"properties": {
"course_id": {
"type": "string",
"description": "Canvas course ID (optional - if not provided, gets announcements for all courses)",
},
},
"required": [],
},
),
types.Tool(
name="get_course_activity_stream",
description="Get recent activity across all courses",
inputSchema={
"type": "object",
"properties": {},
"required": [],
},
),
]
@self.server.call_tool()
async def handle_call_tool(
name: str, arguments: dict[str, Any] | None
) -> list[types.TextContent]:
"""Handle tool calls."""
if not self.canvas_api:
raise RuntimeError("Canvas API not initialized")
try:
if name == "get_courses":
result = await self.canvas_api.get_courses()
elif name == "get_upcoming_assignments":
days_ahead = arguments.get("days_ahead", 30) if arguments else 30
result = await self.canvas_api.get_upcoming_events(days_ahead)
elif name == "get_course_assignments":
if not arguments or "course_id" not in arguments:
raise ValueError("course_id is required")
result = await self.canvas_api.get_course_assignments(arguments["course_id"])
elif name == "get_todo_items":
result = await self.canvas_api.get_todo_items()
elif name == "get_assignment_details":
if not arguments or "course_id" not in arguments or "assignment_id" not in arguments:
raise ValueError("course_id and assignment_id are required")
result = await self.canvas_api.get_assignment_details(
arguments["course_id"], arguments["assignment_id"]
)
elif name == "get_grades":
course_id = arguments.get("course_id") if arguments else None
result = await self.canvas_api.get_grades(course_id)
elif name == "get_calendar_events":
start_date = arguments.get("start_date") if arguments else None
end_date = arguments.get("end_date") if arguments else None
result = await self.canvas_api.get_calendar_events(start_date, end_date)
elif name == "get_announcements":
course_id = arguments.get("course_id") if arguments else None
result = await self.canvas_api.get_announcements(course_id)
elif name == "get_course_activity_stream":
result = await self.canvas_api.get_course_activity_stream()
else:
raise ValueError(f"Unknown tool: {name}")
return [
types.TextContent(
type="text",
text=json.dumps(result, indent=2, default=str),
)
]
except Exception as e:
return [
types.TextContent(
type="text",
text=f"Error executing {name}: {str(e)}",
)
]
async def main():
"""Main entry point for the Canvas MCP server."""
import sys
# Create and setup server
canvas_server = CanvasMCPServer()
canvas_server.setup_handlers()
# Initialize Canvas API
await canvas_server.initialize_canvas_api()
async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
print("🚀 Canvas MCP Server started", file=sys.stderr)
await canvas_server.server.run(
read_stream,
write_stream,
InitializationOptions(
server_name="canvas-mcp-server",
server_version="0.1.0",
capabilities=canvas_server.server.get_capabilities(
notification_options=NotificationOptions(),
experimental_capabilities={},
),
),
)
if __name__ == "__main__":
import sys
asyncio.run(main())
