mpm/odoo-mpm-plugin
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 2 Wiki Activity

feat: add eLearning module (v0.3.0) — 22 new tools

Browse Source
This commit is contained in:
mpmedia
2026-05-28 14:48:25 -05:00
parent 0f60ba2119
commit 8095c24782
1 changed files with 765 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/odoo_mcp.py
+765 -2
View File
@@ -1,13 +1,14 @@
#!/usr/bin/env python3
"""
Odoo MCP Server for MPM
Odoo MCP Server for MPM — v0.3.0
Connects to mpmedia.odoo.com via XML-RPC and exposes tools for
Products, Knowledge, Contacts, Sales, CRM, Project, Helpdesk,
Purchase, Inventory, Employees, and Knowledge Templates.
Purchase, Inventory, Employees, Knowledge Templates, and eLearning.
"""
import os
import re
import base64
import xmlrpc.client
import urllib.request
import urllib.error
@@ -127,6 +128,44 @@ def _create(model, vals):
def _write(model, ids, vals):
return _call(model, "write", [[ids] if isinstance(ids, int) else ids, vals])
def _fetch_image_base64(url: str) -> str:
"""Fetch an image from a URL and return a base64-encoded string."""
try:
req = urllib.request.Request(url, headers={"User-Agent": "odoo-mpm/1.0"})
with urllib.request.urlopen(req, timeout=15) as resp:
return base64.b64encode(resp.read()).decode("utf-8")
except Exception as e:
raise Exception(f"[image fetch] Could not fetch image from {url}: {e}")
def _resolve_partner_ids(emails: list) -> tuple:
"""Resolve a list of email addresses to partner IDs.
Returns (found_ids, not_found_emails)."""
if not emails:
return [], []
lower_emails = [e.lower() for e in emails]
results = _search_read("res.partner",
[["email", "in", lower_emails]],
["id", "email"], limit=len(emails) + 10)
found_map = {r["email"].lower(): r["id"] for r in results if r.get("email")}
found_ids = [found_map[e] for e in lower_emails if e in found_map]
not_found = [e for e in lower_emails if e not in found_map]
return found_ids, not_found
def _resolve_group_ids(group_names: list) -> list:
"""Resolve Odoo group names to IDs (case-insensitive exact match)."""
if not group_names:
return []
group_ids = []
for gname in group_names:
groups = _search_read("res.groups", [["name", "ilike", gname]],
["id", "name"], limit=10)
exact = [g for g in groups if g["name"].strip().lower() == gname.strip().lower()]
if exact:
group_ids.append(exact[0]["id"])
elif groups:
group_ids.append(groups[0]["id"])
return group_ids
# ── FastMCP App ───────────────────────────────────────────────────────────────
mcp = FastMCP("Odoo MPM")
@@ -998,5 +1037,729 @@ def call_odoo_method(model: str, method: str, record_ids: list,
raise Exception(f"[call_odoo_method] Failed calling '{model}.{method}': {e}")
# ════════════════════════════════════════════════════════════════════════════
# ELEARNING — COURSES & CONTENT
# ════════════════════════════════════════════════════════════════════════════
#
# Models:
# slide.channel — Courses
# slide.slide — Slides / content items
# slide.question — Quiz questions (linked to slides)
# slide.answer — Quiz answer options
# slide.channel.partner — Enrollment & completion tracking
# slide.channel.resource — Course resource attachments
#
# Google Drive / image operations:
# Always use the Google Workspace MCP connector when it is available in the
# session. Only ask the user to provide a URL or upload content manually if
# the connector is NOT connected. Never construct Drive URLs manually.
#
# slide_type values:
# 'pdf' — PDF document (url = Drive share link or direct PDF URL)
# 'youtube_video' — YouTube video (url = YouTube watch URL)
# 'vimeo_video' — Vimeo video (url = Vimeo video URL)
# 'infographic' — Image / infographic
# 'webpage' — Article with native HTML content (html_content field)
#
# visibility values: 'public' | 'connected' | 'members' | 'website'
# enroll values: 'public' (Open) | 'invite' (On Invitation)
# channel_type values: 'training' | 'documentation'
# ── Courses ───────────────────────────────────────────────────────────────────
@mcp.tool()
def search_courses(query: str = "", channel_type: str = "", visibility: str = "",
enroll: str = "", published: bool = None, limit: int = 20) -> list:
"""Search eLearning courses.
channel_type : 'training' or 'documentation'
visibility : 'public' (Everyone), 'connected' (Signed In),
'members' (Course Attendees), 'website' (Anyone with link)
enroll : 'public' (Open) or 'invite' (On Invitation)
published : True = live courses only, False = drafts only, None = all
Returns id, name, channel_type, visibility, enroll policy, publish status,
slide counts by type, member count, and website URL.
Trigger phrases: "list courses", "find course", "search elearning", "training courses",
"what courses do we have"."""
domain = []
if query:
domain.append(["name", "ilike", query])
if channel_type:
domain.append(["channel_type", "=", channel_type])
if visibility:
domain.append(["visibility", "=", visibility])
if enroll:
domain.append(["enroll", "=", enroll])
if published is not None:
domain.append(["website_published", "=", published])
return _search_read("slide.channel", domain,
["id", "name", "channel_type", "visibility", "enroll", "website_published",
"total_slides", "nbr_document", "nbr_video", "nbr_infographic",
"members_count", "website_url", "tag_ids"],
limit=limit, order="name asc")
@mcp.tool()
def get_course(channel_id: int) -> dict:
"""Get full details of an eLearning course: settings, description, slide list,
enrollment count, upload/auto-enroll groups, and cover image presence.
Trigger phrases: "get course", "course details", "show course"."""
r = _read("slide.channel", [channel_id],
["id", "name", "description", "channel_type", "visibility", "enroll",
"website_published", "total_slides", "nbr_document", "nbr_video",
"nbr_infographic", "members_count", "website_url", "tag_ids",
"slide_ids", "upload_group_ids", "enroll_group_ids"])
if not r:
return {}
course = r[0]
if course.get("description"):
course["description"] = re.sub(
r'src="data:image/[^;]+;base64,[A-Za-z0-9+/=]+"',
'src="[embedded image]"',
course["description"]
)
return course
@mcp.tool()
def create_course(name: str, description: str = "",
channel_type: str = "training",
visibility: str = "public",
enroll: str = "public",
tag_ids: list = None,
cover_image_url: str = "",
enroll_group_names: list = None) -> dict:
"""Create a new eLearning course.
channel_type : 'training' (default) or 'documentation'
visibility : 'public', 'connected', 'members', 'website'
enroll : 'public' (Open) or 'invite' (On Invitation)
tag_ids : list of existing tag IDs to assign
cover_image_url : image URL — use Google Workspace MCP to get a Drive download
URL when available; only ask the user if the connector is absent
enroll_group_names: Odoo group names for auto-enrollment
e.g. ['Internal Users'] to restrict to employees
Returns dict with new course id and name.
Trigger phrases: "create course", "new course", "new training", "add elearning course"."""
vals = {
"name": name,
"channel_type": channel_type,
"visibility": visibility,
"enroll": enroll,
}
if description:
vals["description"] = description
if tag_ids:
vals["tag_ids"] = [(6, 0, tag_ids)]
if enroll_group_names:
gids = _resolve_group_ids(enroll_group_names)
if gids:
vals["enroll_group_ids"] = [(6, 0, gids)]
channel_id = _create("slide.channel", vals)
if cover_image_url:
try:
_write("slide.channel", channel_id, {"image_1920": _fetch_image_base64(cover_image_url)})
except Exception as e:
return {"id": channel_id, "name": name,
"warning": f"Course created but cover image failed: {e}"}
return {"id": channel_id, "name": name}
@mcp.tool()
def update_course(channel_id: int, name: str = "", description: str = "",
channel_type: str = "", visibility: str = "",
enroll: str = "", tag_ids: list = None,
cover_image_url: str = "",
enroll_group_names: list = None) -> dict:
"""Update an existing eLearning course. Only provided fields are changed.
channel_type : 'training' or 'documentation'
visibility : 'public', 'connected', 'members', 'website'
enroll : 'public' or 'invite'
tag_ids : replaces current tags with this list
cover_image_url : new cover image — use Google Workspace MCP connector when
available to get the download URL; do not ask the user to
construct or upload unless the connector is absent
enroll_group_names: replaces auto-enroll groups; pass [] to clear
Trigger phrases: "update course", "edit course", "change course settings"."""
vals = {}
if name: vals["name"] = name
if description: vals["description"] = description
if channel_type: vals["channel_type"] = channel_type
if visibility: vals["visibility"] = visibility
if enroll: vals["enroll"] = enroll
if tag_ids is not None:
vals["tag_ids"] = [(6, 0, tag_ids)]
if enroll_group_names is not None:
gids = _resolve_group_ids(enroll_group_names)
vals["enroll_group_ids"] = [(6, 0, gids)]
warnings = []
if vals:
_write("slide.channel", channel_id, vals)
if cover_image_url:
try:
_write("slide.channel", channel_id,
{"image_1920": _fetch_image_base64(cover_image_url)})
vals["image_1920"] = "(updated)"
except Exception as e:
warnings.append(f"Cover image update failed: {e}")
result = {"success": True, "channel_id": channel_id,
"fields_updated": list(vals.keys())}
if warnings:
result["warnings"] = warnings
return result
@mcp.tool()
def publish_course(channel_id: int, published: bool = True) -> dict:
"""Publish or unpublish an eLearning course.
published=True → course is live and visible on the website
published=False → course is hidden (draft)
Trigger phrases: "publish course", "unpublish course", "make course live",
"hide course", "take course offline"."""
_write("slide.channel", channel_id, {"website_published": published})
return {"success": True, "channel_id": channel_id,
"status": "published" if published else "unpublished"}
# ── Slides / Content ──────────────────────────────────────────────────────────
@mcp.tool()
def list_course_slides(channel_id: int) -> list:
"""List all slides/content items in a course ordered by sequence.
Returns id, name, slide_type, published status, completion time, sequence, and URL.
Trigger phrases: "list slides", "course content", "course lessons", "show modules"."""
return _search_read("slide.slide",
[["channel_id", "=", channel_id]],
["id", "name", "slide_type", "is_published", "website_published",
"completion_time", "sequence", "url", "tag_ids"],
limit=500, order="sequence asc")
@mcp.tool()
def get_slide(slide_id: int) -> dict:
"""Get full details of a slide including HTML body (articles), URL (pdf/video),
and quiz questions if any are attached.
Trigger phrases: "get slide", "slide details", "show lesson", "get article content"."""
r = _read("slide.slide", [slide_id],
["id", "name", "channel_id", "slide_type", "is_published", "website_published",
"description", "html_content", "url", "completion_time", "sequence",
"tag_ids", "question_ids", "likes", "dislikes"])
if not r:
return {}
slide = r[0]
if slide.get("question_ids"):
try:
questions = _read("slide.question", slide["question_ids"],
["id", "question", "answer_ids", "sequence"])
for q in questions:
if q.get("answer_ids"):
q["answers"] = _read("slide.answer", q["answer_ids"],
["id", "text_value", "is_correct", "comment"])
slide["questions"] = sorted(questions, key=lambda q: q.get("sequence", 0))
except Exception:
slide["questions"] = []
return slide
@mcp.tool()
def create_slide(channel_id: int, name: str,
slide_type: str = "pdf",
description: str = "",
url: str = "",
html_content: str = "",
completion_time: float = 0.0,
sequence: int = 0,
is_published: bool = False) -> dict:
"""Create a new slide/content item in a course.
slide_type options:
'pdf' — PDF document. Set url to a Google Drive share link.
Use Google Workspace MCP to get the link — do not ask
the user to construct it unless the connector is absent.
'youtube_video' — YouTube video. Set url to the watch URL.
'vimeo_video' — Vimeo video. Set url to the Vimeo URL.
'infographic' — Image. Set url to a publicly accessible image URL.
'webpage' — Article. Set html_content to formatted HTML.
Claude can generate article HTML content directly.
url : Link to content. For Drive PDFs, use Google Workspace MCP
to get the shareable link — never ask the user to build it.
html_content : For 'webpage' articles only. Full HTML body.
completion_time : Estimated time in hours (0.25 = 15 min, 0.5 = 30 min).
sequence : Position in course (lower = earlier). 0 = append at end.
is_published : Immediately visible to learners if True.
Trigger phrases: "add slide", "add lesson", "add content", "create slide",
"add video", "add article", "add PDF", "add quiz"."""
vals = {
"channel_id": channel_id,
"name": name,
"slide_type": slide_type,
"is_published": is_published,
"website_published": is_published,
}
if description: vals["description"] = description
if url: vals["url"] = url
if html_content: vals["html_content"] = html_content
if completion_time: vals["completion_time"] = completion_time
if sequence: vals["sequence"] = sequence
slide_id = _create("slide.slide", vals)
return {"id": slide_id, "name": name,
"slide_type": slide_type, "channel_id": channel_id}
@mcp.tool()
def update_slide(slide_id: int, name: str = "", description: str = "",
url: str = "", html_content: str = "",
completion_time: float = None,
sequence: int = None) -> dict:
"""Update a slide's metadata, URL, or article HTML content.
Only provided fields are changed.
For 'webpage' articles: pass html_content to replace the article body.
For pdf/video slides: pass url to update the linked content.
When updating a URL for a Drive document, use Google Workspace MCP to get
the shareable link — do not ask the user to provide it unless the connector
is absent.
Trigger phrases: "update slide", "edit lesson", "update article",
"update slide content", "change video URL"."""
vals = {}
if name: vals["name"] = name
if description: vals["description"] = description
if url: vals["url"] = url
if html_content: vals["html_content"] = html_content
if completion_time is not None: vals["completion_time"] = completion_time
if sequence is not None: vals["sequence"] = sequence
if not vals:
return {"success": False, "reason": "No fields provided to update"}
_write("slide.slide", slide_id, vals)
return {"success": True, "slide_id": slide_id,
"fields_updated": list(vals.keys())}
@mcp.tool()
def publish_slide(slide_id: int, published: bool = True) -> dict:
"""Publish or unpublish an individual slide.
published=True → slide is visible to enrolled learners
published=False → slide is hidden (draft)
Trigger phrases: "publish slide", "unpublish slide", "hide lesson",
"make lesson visible", "publish lesson"."""
_write("slide.slide", slide_id, {
"is_published": published,
"website_published": published,
})
return {"success": True, "slide_id": slide_id,
"status": "published" if published else "unpublished"}
@mcp.tool()
def reorder_slides(channel_id: int, slide_sequences: dict) -> dict:
"""Reorder slides in a course by setting sequence numbers.
slide_sequences maps slide_id → sequence number (lower = earlier).
Example: reorder_slides(8, {"38": 1, "39": 2, "40": 3, "41": 4})
Trigger phrases: "reorder slides", "reorder lessons", "change slide order",
"move lesson", "rearrange course"."""
if not slide_sequences:
return {"success": False, "reason": "No sequences provided"}
updated, errors = [], []
for sid, seq in slide_sequences.items():
try:
_write("slide.slide", int(sid), {"sequence": int(seq)})
updated.append(int(sid))
except Exception as e:
errors.append({"slide_id": sid, "error": str(e)})
return {
"success": len(errors) == 0,
"channel_id": channel_id,
"updated_count": len(updated),
"updated_ids": updated,
"errors": errors,
}
# ── Quiz ──────────────────────────────────────────────────────────────────────
@mcp.tool()
def get_quiz_questions(slide_id: int) -> list:
"""Get all quiz questions and answers for a slide.
Returns each question with its answer options and correct answer flags.
Trigger phrases: "quiz questions", "get questions", "show quiz", "what questions"."""
slide = _read("slide.slide", [slide_id], ["id", "name", "question_ids"])
if not slide or not slide[0].get("question_ids"):
return []
try:
questions = _read("slide.question", slide[0]["question_ids"],
["id", "question", "answer_ids", "sequence"])
for q in questions:
if q.get("answer_ids"):
q["answers"] = _read("slide.answer", q["answer_ids"],
["id", "text_value", "is_correct", "comment"])
return sorted(questions, key=lambda q: q.get("sequence", 0))
except Exception as e:
raise Exception(f"[get_quiz_questions] Failed: {e}")
@mcp.tool()
def add_quiz_question(slide_id: int, question: str,
answers: list, sequence: int = 0) -> dict:
"""Add one quiz question to a slide.
answers: list of dicts with:
'text' — answer option text (required)
'is_correct' — True/False; at least one answer must be correct
'comment' — optional feedback shown after answering
Example:
add_quiz_question(
slide_id=38,
question="What does RDMC stand for?",
answers=[
{"text": "Remote Display Management Console", "is_correct": True,
"comment": "Correct — RDMC manages the full installed display base."},
{"text": "Real-time Data Management Center", "is_correct": False},
{"text": "Remote Display Media Controller", "is_correct": False},
]
)
Trigger phrases: "add question", "add quiz question", "add quiz item"."""
q_vals = {"slide_id": slide_id, "question": question}
if sequence:
q_vals["sequence"] = sequence
question_id = _create("slide.question", q_vals)
answer_ids = []
for ans in answers:
a_vals = {
"question_id": question_id,
"text_value": ans["text"],
"is_correct": ans.get("is_correct", False),
}
if ans.get("comment"):
a_vals["comment"] = ans["comment"]
answer_ids.append(_create("slide.answer", a_vals))
return {
"question_id": question_id,
"slide_id": slide_id,
"question": question,
"answer_ids": answer_ids,
"answer_count": len(answer_ids),
}
@mcp.tool()
def generate_quiz(slide_id: int, questions: list) -> dict:
"""Write a batch of pre-generated quiz questions to a slide.
Intended workflow — Claude does the generation, this tool does the writing:
1. Call get_slide or list_course_slides to read the lesson content
2. Generate question/answer sets based on that content
3. Call this tool to write all questions in one operation
questions: list of dicts with:
'question' — question text (required)
'answers' — list of answer dicts (same format as add_quiz_question)
'sequence' — optional ordering (auto-increments from 1 if omitted)
Replaces any existing questions on the slide before writing.
Trigger phrases: "generate quiz", "create quiz from content",
"auto-generate questions", "build quiz"."""
# Clear existing questions
existing = _read("slide.slide", [slide_id], ["question_ids"])
if existing and existing[0].get("question_ids"):
try:
_call("slide.question", "unlink", [existing[0]["question_ids"]])
except Exception:
pass # non-fatal
created = []
for i, q_data in enumerate(questions):
result = add_quiz_question(
slide_id=slide_id,
question=q_data["question"],
answers=q_data["answers"],
sequence=q_data.get("sequence", i + 1),
)
created.append(result)
return {
"success": True,
"slide_id": slide_id,
"questions_created": len(created),
"question_ids": [c["question_id"] for c in created],
}
# ── Enrollment ────────────────────────────────────────────────────────────────
@mcp.tool()
def get_course_enrollment(channel_id: int) -> list:
"""Get all enrollment records for a course — enrolled users,
completion percentage, and whether they have fully completed it.
Trigger phrases: "who's enrolled", "enrollment", "completion rate",
"course progress", "learner progress"."""
return _search_read("slide.channel.partner",
[["channel_id", "=", channel_id]],
["partner_id", "completion", "completed", "last_seen_slide_id"],
limit=500, order="completion desc")
@mcp.tool()
def enroll_in_course(channel_id: int, partner_ids: list = None,
emails: list = None,
send_invitation: bool = False,
enroll_group_names: list = None) -> dict:
"""Enroll one or more people in an eLearning course.
partner_ids : list of Odoo partner IDs to enroll directly
emails : list of email addresses — resolved to partner IDs automatically
send_invitation : True to send invitation email after enrolling
enroll_group_names: add Odoo group(s) to the course auto-enroll list
(e.g. ['Internal Users'] for all employees)
This tool is the standard onboarding hook — the project management plugin
or any other skill can call it to enroll new customers in courses automatically.
Trigger phrases: "enroll in course", "add to course", "give course access",
"onboard to course", "invite to training"."""
all_partner_ids = list(partner_ids or [])
not_found_emails = []
if emails:
found, not_found = _resolve_partner_ids(emails)
all_partner_ids.extend(found)
not_found_emails = not_found
# Check for existing enrollments to avoid duplicates
existing = _search_read("slide.channel.partner",
[["channel_id", "=", channel_id],
["partner_id", "in", all_partner_ids]],
["partner_id"], limit=1000) if all_partner_ids else []
already_enrolled = set()
for r in existing:
pid = r["partner_id"]
already_enrolled.add(pid[0] if isinstance(pid, (list, tuple)) else pid)
new_ids = [pid for pid in all_partner_ids if pid not in already_enrolled]
# Create enrollment records
created = []
for pid in new_ids:
try:
_create("slide.channel.partner", {
"channel_id": channel_id,
"partner_id": pid,
})
created.append(pid)
except Exception:
pass # may already exist due to race condition
# Add auto-enroll groups if requested
if enroll_group_names:
gids = _resolve_group_ids(enroll_group_names)
if gids:
_write("slide.channel", channel_id,
{"enroll_group_ids": [(4, gid) for gid in gids]})
# Send invitations
invitation_result = None
if send_invitation and created:
for method_name in ("action_send_share_mail", "_send_share_email"):
try:
_call("slide.channel", method_name,
[[channel_id]], {"partner_ids": created})
invitation_result = {"sent": True, "partner_ids": created}
break
except Exception as e:
invitation_result = {"sent": False, "error": str(e)}
result = {
"channel_id": channel_id,
"newly_enrolled": created,
"already_enrolled": list(already_enrolled),
"not_found_emails": not_found_emails,
}
if invitation_result:
result["invitations"] = invitation_result
return result
@mcp.tool()
def bulk_enroll(enrollments: list) -> dict:
"""Enroll multiple people across one or more courses in a single call.
Designed for CSV-style onboarding imports and automated PM → eLearning workflows.
enrollments: list of dicts, each with:
'email' or 'partner_id' — who to enroll (required)
'channel_ids' — list of course IDs (required)
'send_invitation' — optional bool, default False
Example:
bulk_enroll([
{"email": "ops@transitco.gov", "channel_ids": [8, 14], "send_invitation": True},
{"email": "admin@city.gov", "channel_ids": [8], "send_invitation": True},
{"partner_id": 4271, "channel_ids": [15]},
])
Returns a per-record summary with enrolled / already_enrolled / not_found flags.
Trigger phrases: "bulk enroll", "import enrollments", "enroll list",
"onboard customers", "add multiple people to course"."""
results = []
for entry in enrollments:
email = entry.get("email", "")
pid = entry.get("partner_id")
channel_ids = entry.get("channel_ids", [])
send_inv = entry.get("send_invitation", False)
if not channel_ids:
results.append({"entry": entry, "error": "No channel_ids provided"})
continue
for cid in channel_ids:
try:
r = enroll_in_course(
channel_id=cid,
partner_ids=[pid] if pid else None,
emails=[email] if email else None,
send_invitation=send_inv,
)
results.append({
"identity": email or str(pid),
"channel_id": cid,
"enrolled": len(r.get("newly_enrolled", [])) > 0,
"already_enrolled": len(r.get("already_enrolled", [])) > 0,
"not_found": len(r.get("not_found_emails", [])) > 0,
})
except Exception as e:
results.append({
"identity": email or str(pid),
"channel_id": cid,
"error": str(e),
})
succeeded = sum(1 for r in results
if r.get("enrolled") or r.get("already_enrolled"))
return {
"total": len(results),
"succeeded": succeeded,
"failed": len(results) - succeeded,
"results": results,
}
@mcp.tool()
def send_course_invitation(channel_id: int, partner_ids: list) -> dict:
"""Send or resend course invitation emails to specific enrolled partners.
Trigger phrases: "send invitation", "resend invite",
"email course invite", "resend course email"."""
for method_name in ("action_send_share_mail", "_send_share_email"):
try:
_call("slide.channel", method_name,
[[channel_id]], {"partner_ids": partner_ids})
return {"success": True, "channel_id": channel_id,
"invited_partner_ids": partner_ids, "count": len(partner_ids)}
except Exception as e:
last_error = e
raise Exception(f"[send_course_invitation] All invitation methods failed: {last_error}")
# ── Media & Resources ─────────────────────────────────────────────────────────
@mcp.tool()
def set_course_cover(channel_id: int, image_url: str) -> dict:
"""Set or replace the cover image for an eLearning course.
image_url: A direct download URL for the image.
Use Google Workspace MCP to get a download URL from Google Drive
when the connector is available — do not ask the user to provide
or construct the URL unless the connector is absent.
The image is fetched, base64-encoded, and written to Odoo.
Trigger phrases: "set course cover", "update cover image", "course thumbnail",
"change course image"."""
try:
_write("slide.channel", channel_id,
{"image_1920": _fetch_image_base64(image_url)})
return {"success": True, "channel_id": channel_id}
except Exception as e:
raise Exception(f"[set_course_cover] {e}")
@mcp.tool()
def set_slide_cover(slide_id: int, image_url: str) -> dict:
"""Set or replace the thumbnail/cover image for an individual slide.
image_url: A direct download URL for the image.
Use Google Workspace MCP for Drive images when available.
Do not ask the user to provide the URL unless the connector is absent.
Trigger phrases: "slide thumbnail", "slide cover image", "lesson image",
"set slide image"."""
try:
_write("slide.slide", slide_id,
{"image_1920": _fetch_image_base64(image_url)})
return {"success": True, "slide_id": slide_id}
except Exception as e:
raise Exception(f"[set_slide_cover] {e}")
@mcp.tool()
def link_drive_document(slide_id: int, drive_url: str) -> dict:
"""Set the URL for a document, video, or infographic slide.
Use the Google Workspace MCP connector to get the shareable link for
a Drive file when available. Do not ask the user to construct or copy
the URL unless the connector is not connected.
drive_url: Share link or direct viewer URL for the content.
For PDFs: use the Drive file share link.
For videos: use the YouTube or Vimeo watch URL.
Trigger phrases: "link document", "set PDF URL", "attach Google Drive",
"link Drive file", "update video URL", "set slide URL"."""
_write("slide.slide", slide_id, {"url": drive_url})
return {"success": True, "slide_id": slide_id, "url": drive_url}
@mcp.tool()
def add_course_resource(channel_id: int, name: str,
resource_url: str) -> dict:
"""Add a downloadable resource or external link to a course's resource section.
Resources appear as attachments/links on the course landing page.
resource_url: URL of the resource.
Use Google Workspace MCP to get Drive share links when available.
Do not ask the user to provide the URL unless the connector is absent.
Trigger phrases: "add resource", "attach file to course", "course download",
"add course attachment", "add course link"."""
try:
resource_id = _create("slide.channel.resource", {
"channel_id": channel_id,
"name": name,
"resource_type": "url",
"link": resource_url,
})
return {"success": True, "resource_id": resource_id,
"channel_id": channel_id, "name": name, "url": resource_url}
except Exception as e:
# Fallback: add as a published slide if resource model unavailable
try:
sid = _create("slide.slide", {
"channel_id": channel_id,
"name": name,
"slide_type": "pdf",
"url": resource_url,
"is_published": True,
"website_published": True,
})
return {"success": True, "fallback": "created_as_slide",
"slide_id": sid, "channel_id": channel_id,
"name": name, "url": resource_url}
except Exception as e2:
raise Exception(
f"[add_course_resource] Failed: primary={e}, fallback={e2}"
)
if __name__ == "__main__":
mcp.run()