#!/usr/bin/env python3
from __future__ import annotations
import json
import os
from typing import Any, Dict, List, Literal
from google.auth.exceptions import RefreshError
from google.oauth2 import service_account
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
from mcp.server.fastmcp import FastMCP
SCOPE_COURSES_READONLY = "https://www.googleapis.com/auth/classroom.courses.readonly"
SCOPE_COURSES = "https://www.googleapis.com/auth/classroom.courses"
SCOPE_ROSTERS_READONLY = "https://www.googleapis.com/auth/classroom.rosters.readonly"
SCOPE_ROSTERS = "https://www.googleapis.com/auth/classroom.rosters"
SCOPE_PROFILE_EMAILS = "https://www.googleapis.com/auth/classroom.profile.emails"
SCOPE_TOPICS_READONLY = "https://www.googleapis.com/auth/classroom.topics.readonly"
SCOPE_TOPICS = "https://www.googleapis.com/auth/classroom.topics"
SCOPE_ANNOUNCEMENTS = "https://www.googleapis.com/auth/classroom.announcements"
SCOPE_COURSEWORK_STUDENTS_READONLY = (
"https://www.googleapis.com/auth/classroom.coursework.students.readonly"
)
SCOPE_COURSEWORK_STUDENTS = (
"https://www.googleapis.com/auth/classroom.coursework.students"
)
SCOPE_COURSEWORK_ME = "https://www.googleapis.com/auth/classroom.coursework.me"
SCOPE_COURSEWORK_MATERIALS_READONLY = (
"https://www.googleapis.com/auth/classroom.courseworkmaterials.readonly"
)
SCOPE_COURSEWORK_MATERIALS = (
"https://www.googleapis.com/auth/classroom.courseworkmaterials"
)
SERVICE_ACCOUNT_FILE_ENV = "GOOGLE_SERVICE_ACCOUNT_FILE"
DEFAULT_SUBJECT_EMAIL_ENV = "GOOGLE_WORKSPACE_ADMIN_EMAIL"
REQUIRED_TEACHERS_ENV = "CLASSROOM_REQUIRED_TEACHERS"
OBSOLETE_TEACHERS_ENV = "CLASSROOM_OBSOLETE_TEACHERS"
DEFAULT_OBSOLETE_TEACHERS = ["presidencia@asociacionmontessori.mx"]
mcp = FastMCP("google-classroom")
def _required_env(var_name: str) -> str:
value = os.getenv(var_name)
if not value:
raise RuntimeError(
f"Missing required env var '{var_name}'. See .env.example in this folder."
)
return value
def _http_error_to_runtime_error(exc: HttpError) -> RuntimeError:
status = getattr(exc.resp, "status", "unknown")
detail = ""
try:
payload = json.loads(exc.content.decode("utf-8"))
detail = payload.get("error", {}).get("message", "")
except Exception:
detail = str(exc)
message = f"Google Classroom API error (HTTP {status})"
if detail:
message = f"{message}: {detail}"
return RuntimeError(message)
def _refresh_error_to_runtime_error(exc: RefreshError) -> RuntimeError:
return RuntimeError(
"Google OAuth authorization error. "
"Usually this means missing Domain-Wide Delegation scopes for this tool. "
f"Detail: {exc}"
)
def _raise_api_error(exc: Exception) -> None:
if isinstance(exc, HttpError):
raise _http_error_to_runtime_error(exc) from exc
if isinstance(exc, RefreshError):
raise _refresh_error_to_runtime_error(exc) from exc
raise exc
def _classroom_service(
required_scopes: List[str],
acting_user_email: str | None = None,
):
sa_file = _required_env(SERVICE_ACCOUNT_FILE_ENV)
delegated_subject = acting_user_email or _required_env(DEFAULT_SUBJECT_EMAIL_ENV)
creds = service_account.Credentials.from_service_account_file(
sa_file,
scopes=required_scopes,
subject=delegated_subject,
)
return build("classroom", "v1", credentials=creds, cache_discovery=False)
def _to_course_summary(course: Dict[str, Any]) -> Dict[str, Any]:
return {
"id": course.get("id"),
"name": course.get("name"),
"section": course.get("section"),
"descriptionHeading": course.get("descriptionHeading"),
"room": course.get("room"),
"ownerId": course.get("ownerId"),
"courseState": course.get("courseState"),
"creationTime": course.get("creationTime"),
"updateTime": course.get("updateTime"),
"alternateLink": course.get("alternateLink"),
}
def _parse_due_date(date_iso: str | None) -> Dict[str, int] | None:
if not date_iso:
return None
try:
year_str, month_str, day_str = date_iso.split("-")
return {
"year": int(year_str),
"month": int(month_str),
"day": int(day_str),
}
except Exception as exc:
raise ValueError("due_date_iso must be YYYY-MM-DD") from exc
def _parse_due_time(time_hhmm: str | None) -> Dict[str, int] | None:
if not time_hhmm:
return None
try:
hours_str, minutes_str = time_hhmm.split(":")
return {"hours": int(hours_str), "minutes": int(minutes_str)}
except Exception as exc:
raise ValueError("due_time_24h must be HH:MM") from exc
def _normalize_email_list(emails: List[str] | None) -> List[str]:
if not emails:
return []
normalized: List[str] = []
seen: set[str] = set()
for value in emails:
email = (value or "").strip().lower()
if not email or email in seen:
continue
seen.add(email)
normalized.append(email)
return normalized
def _get_required_teacher_emails() -> List[str]:
raw = os.getenv(REQUIRED_TEACHERS_ENV, "")
if not raw.strip():
return []
normalized = raw.replace(";", ",").replace("\n", ",")
return _normalize_email_list([part for part in normalized.split(",") if part.strip()])
def _get_obsolete_teacher_emails() -> List[str]:
raw = os.getenv(OBSOLETE_TEACHERS_ENV)
if raw is None:
return list(DEFAULT_OBSOLETE_TEACHERS)
if not raw.strip():
return []
normalized = raw.replace(";", ",").replace("\n", ",")
return _normalize_email_list([part for part in normalized.split(",") if part.strip()])
def _collect_course_ids_for_teacher(
service: Any,
teacher_email: str,
course_state: str | None = None,
) -> set[str]:
course_ids: set[str] = set()
page_token = None
while True:
kwargs: Dict[str, Any] = {
"teacherId": teacher_email,
"pageToken": page_token,
"pageSize": 100,
}
if course_state:
kwargs["courseStates"] = [course_state]
resp = service.courses().list(**kwargs).execute()
for course in resp.get("courses", []):
course_id = course.get("id")
if course_id:
course_ids.add(course_id)
page_token = resp.get("nextPageToken")
if not page_token:
break
return course_ids
def _collect_obsolete_course_ids(
service: Any,
course_state: str | None = None,
) -> tuple[set[str], List[str]]:
obsolete_teachers = _get_obsolete_teacher_emails()
if not obsolete_teachers:
return set(), []
obsolete_course_ids: set[str] = set()
for teacher_email in obsolete_teachers:
obsolete_course_ids.update(
_collect_course_ids_for_teacher(
service=service,
teacher_email=teacher_email,
course_state=course_state,
)
)
return obsolete_course_ids, obsolete_teachers
def _apply_course_exclusion(
courses: List[Dict[str, Any]],
excluded_course_ids: set[str],
) -> tuple[List[Dict[str, Any]], int]:
if not excluded_course_ids:
return courses, 0
filtered: List[Dict[str, Any]] = []
excluded_count = 0
for course in courses:
course_id = course.get("id")
if course_id and course_id in excluded_course_ids:
excluded_count += 1
continue
filtered.append(course)
return filtered, excluded_count
def _add_teachers_to_course(
service: Any,
course_id: str,
teacher_emails: List[str],
owner_email: str | None = None,
) -> Dict[str, List[Dict[str, str]]]:
added: List[Dict[str, str]] = []
skipped: List[Dict[str, str]] = []
errors: List[Dict[str, str]] = []
owner_normalized = (owner_email or "").strip().lower()
for teacher_email in _normalize_email_list(teacher_emails):
if owner_normalized and teacher_email == owner_normalized:
skipped.append(
{"teacher_email": teacher_email, "reason": "already_owner"}
)
continue
try:
created_teacher = (
service.courses()
.teachers()
.create(courseId=course_id, body={"userId": teacher_email})
.execute()
)
added.append(
{
"teacher_email": teacher_email,
"user_id": created_teacher.get("userId") or teacher_email,
}
)
except Exception as exc:
message = str(exc)
if ("HTTP 409" in message) or ("already exists" in message.lower()):
skipped.append(
{"teacher_email": teacher_email, "reason": "already_teacher"}
)
continue
errors.append({"teacher_email": teacher_email, "error": message})
return {"added": added, "skipped": skipped, "errors": errors}
def _list_all_topics(
service: Any,
course_id: str,
page_size: int = 100,
) -> List[Dict[str, Any]]:
topics: List[Dict[str, Any]] = []
page_token = None
while True:
resp = (
service.courses()
.topics()
.list(courseId=course_id, pageToken=page_token, pageSize=page_size)
.execute()
)
topics.extend(resp.get("topic", []))
page_token = resp.get("nextPageToken")
if not page_token:
break
return topics
def _list_all_coursework(
service: Any,
course_id: str,
page_size: int = 100,
course_work_state: Literal["PUBLISHED", "DRAFT", "DELETED"] | None = None,
) -> List[Dict[str, Any]]:
coursework: List[Dict[str, Any]] = []
page_token = None
while True:
kwargs: Dict[str, Any] = {
"courseId": course_id,
"pageToken": page_token,
"pageSize": page_size,
}
if course_work_state:
kwargs["courseWorkStates"] = [course_work_state]
resp = service.courses().courseWork().list(**kwargs).execute()
coursework.extend(resp.get("courseWork", []))
page_token = resp.get("nextPageToken")
if not page_token:
break
return coursework
def _list_all_coursework_materials(
service: Any,
course_id: str,
page_size: int = 100,
course_work_material_state: Literal["PUBLISHED", "DRAFT", "DELETED"] | None = None,
) -> List[Dict[str, Any]]:
materials: List[Dict[str, Any]] = []
page_token = None
while True:
kwargs: Dict[str, Any] = {
"courseId": course_id,
"pageToken": page_token,
"pageSize": page_size,
}
if course_work_material_state:
kwargs["courseWorkMaterialStates"] = [course_work_material_state]
resp = service.courses().courseWorkMaterials().list(**kwargs).execute()
materials.extend(resp.get("courseWorkMaterial", []))
page_token = resp.get("nextPageToken")
if not page_token:
break
return materials
def _material_identity(material: Dict[str, Any]) -> str:
link = material.get("link")
if isinstance(link, dict):
url = link.get("url")
if url:
return f"link:{url}"
youtube = material.get("youtubeVideo")
if isinstance(youtube, dict):
yid = youtube.get("id") or youtube.get("alternateLink")
if yid:
return f"youtube:{yid}"
drive_file = material.get("driveFile")
if isinstance(drive_file, dict):
inner = drive_file.get("driveFile")
if isinstance(inner, dict):
fid = inner.get("id") or inner.get("alternateLink")
if fid:
return f"drive:{fid}"
return json.dumps(material, sort_keys=True, ensure_ascii=True)
def _dedupe_materials(materials: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
unique: List[Dict[str, Any]] = []
seen: set[str] = set()
for material in materials:
if not isinstance(material, dict):
continue
key = _material_identity(material)
if key in seen:
continue
seen.add(key)
unique.append(material)
return unique
def _filter_retry_safe_materials(materials: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
safe: List[Dict[str, Any]] = []
for material in materials:
if not isinstance(material, dict):
continue
if "link" in material or "youtubeVideo" in material:
safe.append(material)
return _dedupe_materials(safe)
def _is_attachment_visibility_error(exc: Exception) -> bool:
message = str(exc)
return ("AttachmentNotVisible" in message) or ("not visible to the user" in message)
def _clone_coursework_body(
source: Dict[str, Any],
topic_id_map: Dict[str, str],
force_draft_for_created_work: bool,
) -> Dict[str, Any]:
body: Dict[str, Any] = {
"title": source.get("title") or "Untitled coursework",
"workType": source.get("workType") or "ASSIGNMENT",
"state": (
"DRAFT"
if force_draft_for_created_work
else source.get("state")
if source.get("state") in {"DRAFT", "PUBLISHED"}
else "DRAFT"
),
}
passthrough_fields = [
"description",
"materials",
"maxPoints",
"dueDate",
"dueTime",
"scheduledTime",
"submissionModificationMode",
]
for field in passthrough_fields:
value = source.get(field)
if value is not None:
body[field] = value
if "materials" in body and isinstance(body["materials"], list):
body["materials"] = _dedupe_materials(body["materials"])
source_topic_id = source.get("topicId")
if source_topic_id and source_topic_id in topic_id_map:
body["topicId"] = topic_id_map[source_topic_id]
if body["workType"] == "MULTIPLE_CHOICE_QUESTION":
question = source.get("multipleChoiceQuestion")
if question:
body["multipleChoiceQuestion"] = question
elif body["workType"] == "SHORT_ANSWER_QUESTION":
if "shortAnswerQuestion" in source:
body["shortAnswerQuestion"] = source.get("shortAnswerQuestion") or {}
return body
def _clone_coursework_material_body(
source: Dict[str, Any],
topic_id_map: Dict[str, str],
force_draft_for_created_work: bool,
) -> Dict[str, Any]:
body: Dict[str, Any] = {
"title": source.get("title") or "Untitled material",
"state": (
"DRAFT"
if force_draft_for_created_work
else source.get("state")
if source.get("state") in {"DRAFT", "PUBLISHED"}
else "DRAFT"
),
}
passthrough_fields = [
"description",
"materials",
"scheduledTime",
]
for field in passthrough_fields:
value = source.get(field)
if value is not None:
body[field] = value
if "materials" in body and isinstance(body["materials"], list):
body["materials"] = _dedupe_materials(body["materials"])
source_topic_id = source.get("topicId")
if source_topic_id and source_topic_id in topic_id_map:
body["topicId"] = topic_id_map[source_topic_id]
return body
def _build_course_patch_body_and_mask(
*,
name: str | None = None,
section: str | None = None,
description_heading: str | None = None,
description: str | None = None,
room: str | None = None,
owner_id: str | None = None,
course_state: str | None = None,
) -> tuple[Dict[str, Any], str]:
body: Dict[str, Any] = {}
mask_fields: List[str] = []
if name is not None:
body["name"] = name
mask_fields.append("name")
if section is not None:
body["section"] = section
mask_fields.append("section")
if description_heading is not None:
body["descriptionHeading"] = description_heading
mask_fields.append("descriptionHeading")
if description is not None:
body["description"] = description
mask_fields.append("description")
if room is not None:
body["room"] = room
mask_fields.append("room")
if owner_id is not None:
body["ownerId"] = owner_id
mask_fields.append("ownerId")
if course_state is not None:
body["courseState"] = course_state
mask_fields.append("courseState")
if not mask_fields:
raise ValueError(
"No update fields provided. At least one field must be non-null."
)
return body, ",".join(mask_fields)
def _list_courses_for_role(
service: Any,
role_param: Literal["studentId", "teacherId"],
user_email: str,
course_state: str,
) -> List[Dict[str, Any]]:
courses: List[Dict[str, Any]] = []
page_token = None
while True:
req = service.courses().list(
**{role_param: user_email},
courseStates=[course_state],
pageToken=page_token,
pageSize=100,
)
resp = req.execute()
for course in resp.get("courses", []):
courses.append(_to_course_summary(course))
page_token = resp.get("nextPageToken")
if not page_token:
break
return courses
@mcp.tool()
def classroom_get_user_enrollments(
user_email: str,
course_state: Literal["ACTIVE", "ARCHIVED", "PROVISIONED", "DECLINED"] = "ACTIVE",
acting_user_email: str | None = None,
) -> Dict[str, Any]:
"""Return courses where user appears as student and/or teacher."""
try:
service = _classroom_service(
required_scopes=[SCOPE_COURSES_READONLY],
acting_user_email=acting_user_email,
)
student_courses = _list_courses_for_role(
service=service,
role_param="studentId",
user_email=user_email,
course_state=course_state,
)
teacher_courses = _list_courses_for_role(
service=service,
role_param="teacherId",
user_email=user_email,
course_state=course_state,
)
by_id: Dict[str, Dict[str, Any]] = {}
for course in student_courses:
cid = course.get("id")
if cid:
by_id[cid] = {**course, "roles": ["student"]}
for course in teacher_courses:
cid = course.get("id")
if not cid:
continue
if cid in by_id:
by_id[cid]["roles"].append("teacher")
else:
by_id[cid] = {**course, "roles": ["teacher"]}
merged_courses = list(by_id.values())
raw_counts = {
"as_student": len(student_courses),
"as_teacher": len(teacher_courses),
"unique_total": len(merged_courses),
}
obsolete_course_ids, obsolete_teachers = _collect_obsolete_course_ids(
service=service,
course_state=course_state,
)
filtered_courses, excluded_count = _apply_course_exclusion(
courses=merged_courses,
excluded_course_ids=obsolete_course_ids,
)
filtered_counts = {
"as_student": sum(
1 for course in filtered_courses if "student" in course.get("roles", [])
),
"as_teacher": sum(
1 for course in filtered_courses if "teacher" in course.get("roles", [])
),
"unique_total": len(filtered_courses),
}
return {
"user_email": user_email,
"course_state": course_state,
"acting_user_email": acting_user_email
or os.getenv(DEFAULT_SUBJECT_EMAIL_ENV),
"counts": filtered_counts,
"raw_counts": raw_counts,
"obsolete_filter": {
"enabled": bool(obsolete_teachers),
"teacher_emails": obsolete_teachers,
"excluded_courses_count": excluded_count,
},
"courses": filtered_courses,
}
except Exception as exc:
_raise_api_error(exc)
@mcp.tool()
def classroom_get_course(
course_id: str,
acting_user_email: str | None = None,
) -> Dict[str, Any]:
"""Get one course by ID."""
try:
service = _classroom_service(
required_scopes=[SCOPE_COURSES_READONLY],
acting_user_email=acting_user_email,
)
return service.courses().get(id=course_id).execute()
except Exception as exc:
_raise_api_error(exc)
@mcp.tool()
def classroom_list_courses(
course_state: Literal[
"ACTIVE", "ARCHIVED", "PROVISIONED", "DECLINED", "SUSPENDED"
]
| None = None,
teacher_id: str | None = None,
student_id: str | None = None,
page_size: int = 100,
acting_user_email: str | None = None,
) -> Dict[str, Any]:
"""List courses visible to the acting user with optional filters."""
try:
service = _classroom_service(
required_scopes=[SCOPE_COURSES_READONLY],
acting_user_email=acting_user_email,
)
courses: List[Dict[str, Any]] = []
page_token = None
while True:
kwargs: Dict[str, Any] = {
"pageToken": page_token,
"pageSize": page_size,
}
if course_state:
kwargs["courseStates"] = [course_state]
if teacher_id:
kwargs["teacherId"] = teacher_id
if student_id:
kwargs["studentId"] = student_id
resp = service.courses().list(**kwargs).execute()
courses.extend(resp.get("courses", []))
page_token = resp.get("nextPageToken")
if not page_token:
break
obsolete_course_ids, obsolete_teachers = _collect_obsolete_course_ids(
service=service,
course_state=course_state,
)
filtered_courses, excluded_count = _apply_course_exclusion(
courses=courses,
excluded_course_ids=obsolete_course_ids,
)
return {
"count": len(filtered_courses),
"raw_count": len(courses),
"obsolete_filter": {
"enabled": bool(obsolete_teachers),
"teacher_emails": obsolete_teachers,
"excluded_courses_count": excluded_count,
},
"courses": filtered_courses,
}
except Exception as exc:
_raise_api_error(exc)
@mcp.tool()
def classroom_create_course(
name: str,
section: str | None = None,
description_heading: str | None = None,
description: str | None = None,
room: str | None = None,
owner_id: str | None = None,
course_state: Literal["PROVISIONED", "ACTIVE"] = "PROVISIONED",
acting_user_email: str | None = None,
) -> Dict[str, Any]:
"""Create a new course."""
try:
delegated_subject = acting_user_email or _required_env(DEFAULT_SUBJECT_EMAIL_ENV)
required_teachers = _get_required_teacher_emails()
write_scopes = [SCOPE_COURSES]
if required_teachers:
write_scopes.append(SCOPE_ROSTERS)
service = _classroom_service(
required_scopes=write_scopes,
acting_user_email=delegated_subject,
)
body: Dict[str, Any] = {
"name": name,
"courseState": course_state,
}
if section is not None:
body["section"] = section
if description_heading is not None:
body["descriptionHeading"] = description_heading
if description is not None:
body["description"] = description
if room is not None:
body["room"] = room
owner_email = owner_id or delegated_subject
body["ownerId"] = owner_email
created_course = service.courses().create(body=body).execute()
if required_teachers and created_course.get("id"):
required_result = _add_teachers_to_course(
service=service,
course_id=created_course["id"],
teacher_emails=required_teachers,
owner_email=owner_email,
)
created_course["requiredTeachersPolicy"] = {
"enabled": True,
"required_teachers": required_teachers,
"result": required_result,
}
return created_course
except Exception as exc:
_raise_api_error(exc)
@mcp.tool()
def classroom_update_course(
course_id: str,
name: str | None = None,
section: str | None = None,
description_heading: str | None = None,
description: str | None = None,
room: str | None = None,
owner_id: str | None = None,
course_state: Literal["ACTIVE", "ARCHIVED", "PROVISIONED", "DECLINED"] | None = None,
acting_user_email: str | None = None,
) -> Dict[str, Any]:
"""Patch one course and update only provided fields."""
try:
service = _classroom_service(
required_scopes=[SCOPE_COURSES],
acting_user_email=acting_user_email,
)
body, update_mask = _build_course_patch_body_and_mask(
name=name,
section=section,
description_heading=description_heading,
description=description,
room=room,
owner_id=owner_id,
course_state=course_state,
)
return (
service.courses()
.patch(id=course_id, updateMask=update_mask, body=body)
.execute()
)
except Exception as exc:
_raise_api_error(exc)
@mcp.tool()
def classroom_set_course_state(
course_id: str,
course_state: Literal["ACTIVE", "ARCHIVED", "PROVISIONED", "DECLINED"],
acting_user_email: str | None = None,
) -> Dict[str, Any]:
"""Set one course state."""
try:
return classroom_update_course(
course_id=course_id,
course_state=course_state,
acting_user_email=acting_user_email,
)
except Exception as exc:
_raise_api_error(exc)
@mcp.tool()
def classroom_archive_course(
course_id: str,
acting_user_email: str | None = None,
) -> Dict[str, Any]:
"""Archive one course."""
try:
return classroom_set_course_state(
course_id=course_id,
course_state="ARCHIVED",
acting_user_email=acting_user_email,
)
except Exception as exc:
_raise_api_error(exc)
@mcp.tool()
def classroom_activate_course(
course_id: str,
acting_user_email: str | None = None,
) -> Dict[str, Any]:
"""Activate one course."""
try:
return classroom_set_course_state(
course_id=course_id,
course_state="ACTIVE",
acting_user_email=acting_user_email,
)
except Exception as exc:
_raise_api_error(exc)
@mcp.tool()
def classroom_delete_course(
course_id: str,
acting_user_email: str | None = None,
) -> Dict[str, Any]:
"""Delete one course permanently."""
try:
service = _classroom_service(
required_scopes=[SCOPE_COURSES],
acting_user_email=acting_user_email,
)
service.courses().delete(id=course_id).execute()
return {"course_id": course_id, "deleted": True}
except Exception as exc:
_raise_api_error(exc)
@mcp.tool()
def classroom_list_course_students(
course_id: str,
page_size: int = 100,
acting_user_email: str | None = None,
) -> Dict[str, Any]:
"""List students in a course."""
try:
service = _classroom_service(
required_scopes=[SCOPE_ROSTERS_READONLY, SCOPE_PROFILE_EMAILS],
acting_user_email=acting_user_email,
)
students: List[Dict[str, Any]] = []
page_token = None
while True:
resp = (
service.courses()
.students()
.list(courseId=course_id, pageToken=page_token, pageSize=page_size)
.execute()
)
students.extend(resp.get("students", []))
page_token = resp.get("nextPageToken")
if not page_token:
break
return {"course_id": course_id, "count": len(students), "students": students}
except Exception as exc:
_raise_api_error(exc)
@mcp.tool()
def classroom_list_course_teachers(
course_id: str,
page_size: int = 100,
acting_user_email: str | None = None,
) -> Dict[str, Any]:
"""List teachers in a course."""
try:
service = _classroom_service(
required_scopes=[SCOPE_ROSTERS_READONLY, SCOPE_PROFILE_EMAILS],
acting_user_email=acting_user_email,
)
teachers: List[Dict[str, Any]] = []
page_token = None
while True:
resp = (
service.courses()
.teachers()
.list(courseId=course_id, pageToken=page_token, pageSize=page_size)
.execute()
)
teachers.extend(resp.get("teachers", []))
page_token = resp.get("nextPageToken")
if not page_token:
break
return {"course_id": course_id, "count": len(teachers), "teachers": teachers}
except Exception as exc:
_raise_api_error(exc)
@mcp.tool()
def classroom_add_teacher_to_course(
course_id: str,
teacher_email: str,
acting_user_email: str | None = None,
) -> Dict[str, Any]:
"""Add a teacher to a course by email."""
try:
service = _classroom_service(
required_scopes=[SCOPE_ROSTERS],
acting_user_email=acting_user_email,
)
body = {"userId": teacher_email}
return service.courses().teachers().create(courseId=course_id, body=body).execute()
except Exception as exc:
_raise_api_error(exc)
@mcp.tool()
def classroom_remove_teacher_from_course(
course_id: str,
teacher_email: str,
acting_user_email: str | None = None,
) -> Dict[str, Any]:
"""Remove a teacher from a course by email."""
try:
service = _classroom_service(
required_scopes=[SCOPE_ROSTERS],
acting_user_email=acting_user_email,
)
service.courses().teachers().delete(courseId=course_id, userId=teacher_email).execute()
return {
"course_id": course_id,
"teacher_email": teacher_email,
"removed": True,
}
except Exception as exc:
_raise_api_error(exc)
@mcp.tool()
def classroom_add_student_to_course(
course_id: str,
student_email: str,
acting_user_email: str | None = None,
) -> Dict[str, Any]:
"""Add a student to a course by email."""
try:
service = _classroom_service(
required_scopes=[SCOPE_ROSTERS],
acting_user_email=acting_user_email,
)
body = {"userId": student_email}
return service.courses().students().create(courseId=course_id, body=body).execute()
except Exception as exc:
_raise_api_error(exc)
@mcp.tool()
def classroom_remove_student_from_course(
course_id: str,
student_email: str,
acting_user_email: str | None = None,
) -> Dict[str, Any]:
"""Remove a student from a course by email."""
try:
service = _classroom_service(
required_scopes=[SCOPE_ROSTERS],
acting_user_email=acting_user_email,
)
service.courses().students().delete(courseId=course_id, userId=student_email).execute()
return {
"course_id": course_id,
"student_email": student_email,
"removed": True,
}
except Exception as exc:
_raise_api_error(exc)
@mcp.tool()
def classroom_list_invitations(
course_id: str | None = None,
user_id: str | None = None,
page_size: int = 100,
acting_user_email: str | None = None,
) -> Dict[str, Any]:
"""List invitations with optional course/user filters."""
try:
service = _classroom_service(
required_scopes=[SCOPE_ROSTERS_READONLY],
acting_user_email=acting_user_email,
)
invitations: List[Dict[str, Any]] = []
page_token = None
while True:
kwargs: Dict[str, Any] = {
"pageToken": page_token,
"pageSize": page_size,
}
if course_id:
kwargs["courseId"] = course_id
if user_id:
kwargs["userId"] = user_id
resp = service.invitations().list(**kwargs).execute()
invitations.extend(resp.get("invitations", []))
page_token = resp.get("nextPageToken")
if not page_token:
break
return {"count": len(invitations), "invitations": invitations}
except Exception as exc:
_raise_api_error(exc)
@mcp.tool()
def classroom_get_invitation(
invitation_id: str,
acting_user_email: str | None = None,
) -> Dict[str, Any]:
"""Get one invitation by ID."""
try:
service = _classroom_service(
required_scopes=[SCOPE_ROSTERS_READONLY],
acting_user_email=acting_user_email,
)
return service.invitations().get(id=invitation_id).execute()
except Exception as exc:
_raise_api_error(exc)
@mcp.tool()
def classroom_create_invitation(
course_id: str,
user_id: str,
role: Literal["STUDENT", "TEACHER"],
acting_user_email: str | None = None,
) -> Dict[str, Any]:
"""Create a student/teacher invitation."""
try:
service = _classroom_service(
required_scopes=[SCOPE_ROSTERS],
acting_user_email=acting_user_email,
)
body = {
"courseId": course_id,
"userId": user_id,
"role": role,
}
return service.invitations().create(body=body).execute()
except Exception as exc:
_raise_api_error(exc)
@mcp.tool()
def classroom_delete_invitation(
invitation_id: str,
acting_user_email: str | None = None,
) -> Dict[str, Any]:
"""Delete one invitation."""
try:
service = _classroom_service(
required_scopes=[SCOPE_ROSTERS],
acting_user_email=acting_user_email,
)
service.invitations().delete(id=invitation_id).execute()
return {"invitation_id": invitation_id, "deleted": True}
except Exception as exc:
_raise_api_error(exc)
@mcp.tool()
def classroom_accept_invitation(
invitation_id: str,
acting_user_email: str | None = None,
) -> Dict[str, Any]:
"""Accept one invitation as the invited user."""
try:
service = _classroom_service(
required_scopes=[SCOPE_ROSTERS],
acting_user_email=acting_user_email,
)
return service.invitations().accept(id=invitation_id).execute()
except Exception as exc:
_raise_api_error(exc)
@mcp.tool()
def classroom_list_course_topics(
course_id: str,
page_size: int = 100,
acting_user_email: str | None = None,
) -> Dict[str, Any]:
"""List topics in a course."""
try:
service = _classroom_service(
required_scopes=[SCOPE_TOPICS_READONLY],
acting_user_email=acting_user_email,
)
topics: List[Dict[str, Any]] = []
page_token = None
while True:
resp = (
service.courses()
.topics()
.list(courseId=course_id, pageToken=page_token, pageSize=page_size)
.execute()
)
topics.extend(resp.get("topic", []))
page_token = resp.get("nextPageToken")
if not page_token:
break
return {"course_id": course_id, "count": len(topics), "topics": topics}
except Exception as exc:
_raise_api_error(exc)
@mcp.tool()
def classroom_create_topic(
course_id: str,
name: str,
acting_user_email: str | None = None,
) -> Dict[str, Any]:
"""Create a topic in a course."""
try:
service = _classroom_service(
required_scopes=[SCOPE_TOPICS],
acting_user_email=acting_user_email,
)
body = {"name": name}
return service.courses().topics().create(courseId=course_id, body=body).execute()
except Exception as exc:
_raise_api_error(exc)
@mcp.tool()
def classroom_create_announcement(
course_id: str,
text: str,
state: Literal["PUBLISHED", "DRAFT"] = "PUBLISHED",
topic_id: str | None = None,
acting_user_email: str | None = None,
) -> Dict[str, Any]:
"""Create an announcement in a course."""
try:
service = _classroom_service(
required_scopes=[SCOPE_ANNOUNCEMENTS],
acting_user_email=acting_user_email,
)
body: Dict[str, Any] = {
"courseId": course_id,
"text": text,
"state": state,
}
if topic_id:
# Classroom announcement resource does not support topicId.
body["text"] = f"{text}\n\n[topic:{topic_id}]"
return service.courses().announcements().create(courseId=course_id, body=body).execute()
except Exception as exc:
_raise_api_error(exc)
@mcp.tool()
def classroom_list_coursework_materials(
course_id: str,
page_size: int = 100,
course_work_material_state: Literal["PUBLISHED", "DRAFT", "DELETED"] | None = None,
acting_user_email: str | None = None,
) -> Dict[str, Any]:
"""List course work materials in a course."""
try:
service = _classroom_service(
required_scopes=[SCOPE_COURSEWORK_MATERIALS_READONLY],
acting_user_email=acting_user_email,
)
materials = _list_all_coursework_materials(
service=service,
course_id=course_id,
page_size=page_size,
course_work_material_state=course_work_material_state,
)
return {
"course_id": course_id,
"count": len(materials),
"courseWorkMaterial": materials,
}
except Exception as exc:
_raise_api_error(exc)
@mcp.tool()
def classroom_create_coursework_material(
course_id: str,
title: str,
description: str | None = None,
materials: List[Dict[str, Any]] | None = None,
material_links: List[str] | None = None,
state: Literal["DRAFT", "PUBLISHED"] = "DRAFT",
topic_id: str | None = None,
acting_user_email: str | None = None,
) -> Dict[str, Any]:
"""Create a course work material item."""
try:
service = _classroom_service(
required_scopes=[SCOPE_COURSEWORK_MATERIALS],
acting_user_email=acting_user_email,
)
body: Dict[str, Any] = {
"title": title,
"state": state,
}
if description:
body["description"] = description
if topic_id:
body["topicId"] = topic_id
merged_materials: List[Dict[str, Any]] = []
if materials:
merged_materials.extend(materials)
if material_links:
for url in material_links:
clean_url = (url or "").strip()
if clean_url:
merged_materials.append({"link": {"url": clean_url}})
if merged_materials:
body["materials"] = merged_materials
return (
service.courses()
.courseWorkMaterials()
.create(courseId=course_id, body=body)
.execute()
)
except Exception as exc:
_raise_api_error(exc)
@mcp.tool()
def classroom_list_coursework(
course_id: str,
page_size: int = 100,
course_work_state: Literal["PUBLISHED", "DRAFT", "DELETED"] | None = None,
acting_user_email: str | None = None,
) -> Dict[str, Any]:
"""List coursework in a course."""
try:
service = _classroom_service(
required_scopes=[SCOPE_COURSEWORK_STUDENTS_READONLY],
acting_user_email=acting_user_email,
)
coursework: List[Dict[str, Any]] = []
page_token = None
while True:
kwargs: Dict[str, Any] = {
"courseId": course_id,
"pageToken": page_token,
"pageSize": page_size,
}
if course_work_state:
kwargs["courseWorkStates"] = [course_work_state]
resp = service.courses().courseWork().list(**kwargs).execute()
coursework.extend(resp.get("courseWork", []))
page_token = resp.get("nextPageToken")
if not page_token:
break
return {"course_id": course_id, "count": len(coursework), "courseWork": coursework}
except Exception as exc:
_raise_api_error(exc)
@mcp.tool()
def classroom_create_assignment(
course_id: str,
title: str,
instructions: str | None = None,
max_points: float = 100.0,
state: Literal["DRAFT", "PUBLISHED"] = "DRAFT",
due_date_iso: str | None = None,
due_time_24h: str | None = None,
topic_id: str | None = None,
acting_user_email: str | None = None,
) -> Dict[str, Any]:
"""Create assignment coursework."""
try:
due_date = _parse_due_date(due_date_iso)
due_time = _parse_due_time(due_time_24h)
if due_time and not due_date:
raise ValueError("due_time_24h requires due_date_iso")
service = _classroom_service(
required_scopes=[SCOPE_COURSEWORK_STUDENTS],
acting_user_email=acting_user_email,
)
body: Dict[str, Any] = {
"title": title,
"workType": "ASSIGNMENT",
"state": state,
"maxPoints": max_points,
}
if instructions:
body["description"] = instructions
if due_date:
body["dueDate"] = due_date
if due_time:
body["dueTime"] = due_time
if topic_id:
body["topicId"] = topic_id
return service.courses().courseWork().create(courseId=course_id, body=body).execute()
except Exception as exc:
_raise_api_error(exc)
@mcp.tool()
def classroom_clone_course(
source_course_id: str,
new_course_name: str,
teacher_emails: List[str] | None = None,
student_emails: List[str] | None = None,
new_course_state: Literal["PROVISIONED", "ACTIVE"] = "PROVISIONED",
copy_topics: bool = True,
copy_coursework: bool = True,
copy_coursework_materials: bool = True,
force_draft_for_created_work: bool = True,
acting_user_email: str | None = None,
) -> Dict[str, Any]:
"""Clone a course into a new course and optionally add teachers/students."""
try:
delegated_subject = acting_user_email or _required_env(DEFAULT_SUBJECT_EMAIL_ENV)
required_teachers = _get_required_teacher_emails()
combined_teacher_emails = _normalize_email_list(
(teacher_emails or []) + required_teachers
)
read_scopes = [SCOPE_COURSES_READONLY]
if copy_topics:
read_scopes.append(SCOPE_TOPICS_READONLY)
if copy_coursework:
read_scopes.append(SCOPE_COURSEWORK_STUDENTS_READONLY)
if copy_coursework_materials:
read_scopes.append(SCOPE_COURSEWORK_MATERIALS_READONLY)
write_scopes = [SCOPE_COURSES]
if combined_teacher_emails or student_emails:
write_scopes.append(SCOPE_ROSTERS)
if copy_topics:
write_scopes.append(SCOPE_TOPICS)
if copy_coursework:
write_scopes.append(SCOPE_COURSEWORK_STUDENTS)
if copy_coursework_materials:
write_scopes.append(SCOPE_COURSEWORK_MATERIALS)
read_service = _classroom_service(
required_scopes=read_scopes,
acting_user_email=delegated_subject,
)
write_service = _classroom_service(
required_scopes=write_scopes,
acting_user_email=delegated_subject,
)
source_course = read_service.courses().get(id=source_course_id).execute()
create_body: Dict[str, Any] = {
"name": new_course_name,
"courseState": new_course_state,
"ownerId": delegated_subject,
}
for field in ["section", "descriptionHeading", "description", "room"]:
value = source_course.get(field)
if value is not None:
create_body[field] = value
target_course = write_service.courses().create(body=create_body).execute()
target_course_id = target_course.get("id")
result: Dict[str, Any] = {
"source_course_id": source_course_id,
"target_course": _to_course_summary(target_course),
"settings": {
"copy_topics": copy_topics,
"copy_coursework": copy_coursework,
"copy_coursework_materials": copy_coursework_materials,
"force_draft_for_created_work": force_draft_for_created_work,
"required_teachers": required_teachers,
},
"enrollment": {
"teachers_added": [],
"teachers_skipped": [],
"teachers_errors": [],
"students_added": [],
"students_skipped": [],
"students_errors": [],
},
"topics": {"source_count": 0, "created_count": 0, "errors": []},
"coursework": {
"source_count": 0,
"created_count": 0,
"errors": [],
"warnings": [],
},
"coursework_materials": {
"source_count": 0,
"created_count": 0,
"errors": [],
"warnings": [],
},
"success": True,
}
for teacher_email in combined_teacher_emails:
if teacher_email == delegated_subject.lower():
result["enrollment"]["teachers_skipped"].append(
{
"teacher_email": teacher_email,
"reason": "already_owner",
}
)
continue
try:
created_teacher = (
write_service.courses()
.teachers()
.create(
courseId=target_course_id,
body={"userId": teacher_email},
)
.execute()
)
result["enrollment"]["teachers_added"].append(
created_teacher.get("userId") or teacher_email
)
except Exception as exc:
result["enrollment"]["teachers_errors"].append(
{"teacher_email": teacher_email, "error": str(exc)}
)
normalized_students = _normalize_email_list(student_emails)
for student_email in normalized_students:
try:
created_student = (
write_service.courses()
.students()
.create(
courseId=target_course_id,
body={"userId": student_email},
)
.execute()
)
result["enrollment"]["students_added"].append(
created_student.get("userId") or student_email
)
except Exception as exc:
result["enrollment"]["students_errors"].append(
{"student_email": student_email, "error": str(exc)}
)
topic_id_map: Dict[str, str] = {}
if copy_topics:
source_topics = _list_all_topics(read_service, source_course_id)
result["topics"]["source_count"] = len(source_topics)
for source_topic in source_topics:
source_topic_id = source_topic.get("topicId")
topic_name = source_topic.get("name") or "Untitled topic"
try:
created_topic = (
write_service.courses()
.topics()
.create(
courseId=target_course_id,
body={"name": topic_name},
)
.execute()
)
created_topic_id = created_topic.get("topicId")
if source_topic_id and created_topic_id:
topic_id_map[source_topic_id] = created_topic_id
result["topics"]["created_count"] += 1
except Exception as exc:
result["topics"]["errors"].append(
{
"source_topic_id": source_topic_id,
"topic_name": topic_name,
"error": str(exc),
}
)
if copy_coursework:
source_coursework = _list_all_coursework(read_service, source_course_id)
result["coursework"]["source_count"] = len(source_coursework)
for source_item in source_coursework:
source_item_id = source_item.get("id")
source_title = source_item.get("title") or "Untitled coursework"
body = _clone_coursework_body(
source=source_item,
topic_id_map=topic_id_map,
force_draft_for_created_work=force_draft_for_created_work,
)
try:
created_item = (
write_service.courses()
.courseWork()
.create(courseId=target_course_id, body=body)
.execute()
)
result["coursework"]["created_count"] += 1
result["coursework"].setdefault("items", []).append(
{
"source_id": source_item_id,
"source_title": source_title,
"created_id": created_item.get("id"),
}
)
except Exception as exc:
if _is_attachment_visibility_error(exc) and body.get("materials"):
retry_body = dict(body)
retry_materials = _filter_retry_safe_materials(
retry_body.get("materials", [])
)
if retry_materials:
retry_body["materials"] = retry_materials
else:
retry_body.pop("materials", None)
try:
created_item = (
write_service.courses()
.courseWork()
.create(courseId=target_course_id, body=retry_body)
.execute()
)
result["coursework"]["created_count"] += 1
result["coursework"].setdefault("items", []).append(
{
"source_id": source_item_id,
"source_title": source_title,
"created_id": created_item.get("id"),
}
)
result["coursework"]["warnings"].append(
{
"source_id": source_item_id,
"source_title": source_title,
"warning": (
"Copied with filtered materials due to "
"attachment visibility constraints."
),
}
)
continue
except Exception as retry_exc:
result["coursework"]["errors"].append(
{
"source_id": source_item_id,
"source_title": source_title,
"error": str(retry_exc),
"first_error": str(exc),
}
)
continue
result["coursework"]["errors"].append(
{
"source_id": source_item_id,
"source_title": source_title,
"error": str(exc),
}
)
if copy_coursework_materials:
source_material_items = _list_all_coursework_materials(
read_service, source_course_id
)
result["coursework_materials"]["source_count"] = len(source_material_items)
for source_item in source_material_items:
source_item_id = source_item.get("id")
source_title = source_item.get("title") or "Untitled material"
body = _clone_coursework_material_body(
source=source_item,
topic_id_map=topic_id_map,
force_draft_for_created_work=force_draft_for_created_work,
)
try:
created_item = (
write_service.courses()
.courseWorkMaterials()
.create(courseId=target_course_id, body=body)
.execute()
)
result["coursework_materials"]["created_count"] += 1
result["coursework_materials"].setdefault("items", []).append(
{
"source_id": source_item_id,
"source_title": source_title,
"created_id": created_item.get("id"),
}
)
except Exception as exc:
if _is_attachment_visibility_error(exc) and body.get("materials"):
retry_body = dict(body)
retry_materials = _filter_retry_safe_materials(
retry_body.get("materials", [])
)
if retry_materials:
retry_body["materials"] = retry_materials
else:
retry_body.pop("materials", None)
try:
created_item = (
write_service.courses()
.courseWorkMaterials()
.create(courseId=target_course_id, body=retry_body)
.execute()
)
result["coursework_materials"]["created_count"] += 1
result["coursework_materials"].setdefault("items", []).append(
{
"source_id": source_item_id,
"source_title": source_title,
"created_id": created_item.get("id"),
}
)
result["coursework_materials"]["warnings"].append(
{
"source_id": source_item_id,
"source_title": source_title,
"warning": (
"Copied with filtered materials due to "
"attachment visibility constraints."
),
}
)
continue
except Exception as retry_exc:
result["coursework_materials"]["errors"].append(
{
"source_id": source_item_id,
"source_title": source_title,
"error": str(retry_exc),
"first_error": str(exc),
}
)
continue
result["coursework_materials"]["errors"].append(
{
"source_id": source_item_id,
"source_title": source_title,
"error": str(exc),
}
)
has_errors = any(
[
bool(result["enrollment"]["teachers_errors"]),
bool(result["enrollment"]["students_errors"]),
bool(result["topics"]["errors"]),
bool(result["coursework"]["errors"]),
bool(result["coursework_materials"]["errors"]),
]
)
result["success"] = not has_errors
return result
except Exception as exc:
_raise_api_error(exc)
@mcp.tool()
def classroom_list_student_submissions(
course_id: str,
course_work_id: str,
submission_state: Literal[
"NEW",
"CREATED",
"TURNED_IN",
"RETURNED",
"RECLAIMED_BY_STUDENT",
]
| None = None,
user_id: str | None = None,
page_size: int = 100,
acting_user_email: str | None = None,
) -> Dict[str, Any]:
"""List student submissions for one coursework."""
try:
service = _classroom_service(
required_scopes=[SCOPE_COURSEWORK_STUDENTS_READONLY],
acting_user_email=acting_user_email,
)
submissions: List[Dict[str, Any]] = []
page_token = None
while True:
kwargs: Dict[str, Any] = {
"courseId": course_id,
"courseWorkId": course_work_id,
"pageToken": page_token,
"pageSize": page_size,
}
if submission_state:
kwargs["states"] = [submission_state]
if user_id:
kwargs["userId"] = user_id
resp = (
service.courses()
.courseWork()
.studentSubmissions()
.list(**kwargs)
.execute()
)
submissions.extend(resp.get("studentSubmissions", []))
page_token = resp.get("nextPageToken")
if not page_token:
break
return {
"course_id": course_id,
"course_work_id": course_work_id,
"count": len(submissions),
"studentSubmissions": submissions,
}
except Exception as exc:
_raise_api_error(exc)
@mcp.tool()
def classroom_set_draft_grade(
course_id: str,
course_work_id: str,
submission_id: str,
draft_grade: float,
acting_user_email: str | None = None,
) -> Dict[str, Any]:
"""Set draft grade on one submission."""
try:
service = _classroom_service(
required_scopes=[SCOPE_COURSEWORK_STUDENTS],
acting_user_email=acting_user_email,
)
body = {"draftGrade": draft_grade}
return (
service.courses()
.courseWork()
.studentSubmissions()
.patch(
courseId=course_id,
courseWorkId=course_work_id,
id=submission_id,
updateMask="draftGrade",
body=body,
)
.execute()
)
except Exception as exc:
_raise_api_error(exc)
@mcp.tool()
def classroom_set_assigned_grade(
course_id: str,
course_work_id: str,
submission_id: str,
assigned_grade: float,
acting_user_email: str | None = None,
) -> Dict[str, Any]:
"""Set assigned grade on one submission."""
try:
service = _classroom_service(
required_scopes=[SCOPE_COURSEWORK_STUDENTS],
acting_user_email=acting_user_email,
)
body = {"assignedGrade": assigned_grade}
return (
service.courses()
.courseWork()
.studentSubmissions()
.patch(
courseId=course_id,
courseWorkId=course_work_id,
id=submission_id,
updateMask="assignedGrade",
body=body,
)
.execute()
)
except Exception as exc:
_raise_api_error(exc)
@mcp.tool()
def classroom_turn_in_submission(
course_id: str,
course_work_id: str,
submission_id: str,
acting_user_email: str | None = None,
) -> Dict[str, Any]:
"""Turn in a submission as the student owner."""
try:
service = _classroom_service(
required_scopes=[SCOPE_COURSEWORK_ME],
acting_user_email=acting_user_email,
)
return (
service.courses()
.courseWork()
.studentSubmissions()
.turnIn(
courseId=course_id,
courseWorkId=course_work_id,
id=submission_id,
body={},
)
.execute()
)
except Exception as exc:
_raise_api_error(exc)
@mcp.tool()
def classroom_reclaim_submission(
course_id: str,
course_work_id: str,
submission_id: str,
acting_user_email: str | None = None,
) -> Dict[str, Any]:
"""Reclaim a submitted item as the student owner."""
try:
service = _classroom_service(
required_scopes=[SCOPE_COURSEWORK_ME],
acting_user_email=acting_user_email,
)
return (
service.courses()
.courseWork()
.studentSubmissions()
.reclaim(
courseId=course_id,
courseWorkId=course_work_id,
id=submission_id,
body={},
)
.execute()
)
except Exception as exc:
_raise_api_error(exc)
@mcp.tool()
def classroom_return_submission(
course_id: str,
course_work_id: str,
submission_id: str,
acting_user_email: str | None = None,
) -> Dict[str, Any]:
"""Return a submission to a student."""
try:
service = _classroom_service(
required_scopes=[SCOPE_COURSEWORK_STUDENTS],
acting_user_email=acting_user_email,
)
return (
service.courses()
.courseWork()
.studentSubmissions()
.return_(
courseId=course_id,
courseWorkId=course_work_id,
id=submission_id,
body={},
)
.execute()
)
except Exception as exc:
_raise_api_error(exc)
if __name__ == "__main__":
mcp.run(transport="stdio")
