From 8095c247827adf19ae281ed2800596184acc1015 Mon Sep 17 00:00:00 2001 From: mpmedia Date: Thu, 28 May 2026 14:48:25 -0500 Subject: [PATCH] =?UTF-8?q?feat:=20add=20eLearning=20module=20(v0.3.0)=20?= =?UTF-8?q?=E2=80=94=2022=20new=20tools?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- server/odoo_mcp.py | 767 ++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 765 insertions(+), 2 deletions(-) diff --git a/server/odoo_mcp.py b/server/odoo_mcp.py index 39a85e4..3e04fc3 100644 --- a/server/odoo_mcp.py +++ b/server/odoo_mcp.py @@ -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()