#!/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_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"
SERVICE_ACCOUNT_FILE_ENV = "GOOGLE_SERVICE_ACCOUNT_FILE"
DEFAULT_SUBJECT_EMAIL_ENV = "GOOGLE_WORKSPACE_ADMIN_EMAIL"
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 _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"]}
return {
"user_email": user_email,
"course_state": course_state,
"acting_user_email": acting_user_email
or os.getenv(DEFAULT_SUBJECT_EMAIL_ENV),
"counts": {
"as_student": len(student_courses),
"as_teacher": len(teacher_courses),
"unique_total": len(by_id),
},
"courses": list(by_id.values()),
}
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
return {"count": len(courses), "courses": 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)
service = _classroom_service(
required_scopes=[SCOPE_COURSES],
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
body["ownerId"] = owner_id or delegated_subject
return service.courses().create(body=body).execute()
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],
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],
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(
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_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")
