fork commit

mcp_servers/zoom/README.md

@@ -0,0 +1,55 @@
+# MCP Zoom Server
+
+A Model Context Protocol (MCP) server that exposes Zoom meeting, user, and recording functionality as tools. This allows MCP-compatible clients to interact with Zoom without manual API calls.
+
+## Purpose
+
+The server wraps common Zoom API actions (list, create, update, delete meetings; list users; retrieve recordings) into simple, atomic tools. Each tool returns structured JSON, making it easy for an AI client to chain calls or process the results.
+
+---
+
+## Installation & Setup
+
+1. Clone the repository
+2. Create a virtual environment
+3. Install dependencies
+install "mcp[cli]" httpx pydantic python-dotenv tenacity
+install python-dotenv mcp fastmcp
+4. Get Zoom API Credentials
+Go to Zoom App Marketplace → Build App → choose Server-to-Server OAuth.
+
+Fill in the required details.
+Under Scopes, include at least:
+Meetings: meeting:read:admin, meeting:write:admin
+Users: user:read:admin, user:read:list_users:admin
+Recordings: recording:read:admin
+
+Activate the app and copy:
+Account ID
+Client ID
+Client Secret
+
+Environment Variables
+Store credentials in a .env file in the project root:
+
+ZOOM_ACCOUNT_ID=your_account_id
+ZOOM_CLIENT_ID=your_client_id
+ZOOM_CLIENT_SECRET=your_client_secret
+Do not commit this file to source control.
+
+5. Running the Server
+python -m src.server
+
+
+6. Available Tools
+Tool	Description
+health_check()	Checks server status and required environment variables
+list_users()	Lists Zoom users in the account
+list_meetings(user_id_or_email, type, page_size)	Lists meetings for a specific user
+create_meeting(user_id_or_email, topic, start_time_iso, duration_minutes, timezone, settings)	Creates a scheduled meeting
+get_meeting(meeting_id)	Retrieves meeting details
+update_meeting(meeting_id, patch_fields)	Updates meeting properties
+delete_meeting(meeting_id)	Deletes a meeting
+list_recordings_for_meeting(meeting_id)	Lists recordings for a given meeting
+list_recordings_for_user(user_id_or_email, from, to, page_size)	Lists recordings for a user in a date range
+

mcp_servers/zoom/src/server.py

@@ -0,0 +1,166 @@
+import os, json
+from dotenv import load_dotenv
+from mcp.server.fastmcp import FastMCP
+from src.zoom_client import ZoomClient, ZoomError
+
+# Load environment variables from .env file 
+load_dotenv()
+
+# Initialize the MCP server with name
+mcp = FastMCP("mcp-zoom")
+
+# Cache for the ZoomClient instance
+_zoom: ZoomClient | None = None
+
+def zoom() -> ZoomClient:
+    """Return a singleton ZoomClient instance."""
+    global _zoom
+    if _zoom is None:
+        _zoom = ZoomClient()
+    return _zoom
+
+# Helpers for output formatting
+def _ok(data: dict) -> str:
+    """Wrap a successful result in JSON."""
+    return json.dumps(data, ensure_ascii=False)
+
+def _err(msg: str) -> str:
+    """Wrap an error message in JSON."""
+    return json.dumps({"error": msg}, ensure_ascii=False)
+
+
+# MCP Tools
+
+@mcp.tool()
+def health_check() -> str:
+    """
+    Check if the server is running and if required Zoom credentials are set.
+    Returns 'ok' if all required env vars exist, otherwise lists the missing ones.
+    """
+    missing = [k for k in ["ZOOM_ACCOUNT_ID", "ZOOM_CLIENT_ID", "ZOOM_CLIENT_SECRET"] if not os.getenv(k)]
+    status = "ok" if not missing else f"missing env: {', '.join(missing)}"
+    return _ok({"status": status})
+
+@mcp.tool()
+async def list_meetings(user_id_or_email: str, type: str = "upcoming", page_size: int = 10) -> str:
+    """
+    List Zoom meetings for a specific user (by email or user ID).
+    Defaults to upcoming meetings, limited by page_size.
+    """
+    try:
+        data = await zoom().list_meetings_for_user(user_id_or_email, page_size, type)
+        meetings = [{
+            "id": m["id"],
+            "topic": m.get("topic", ""),
+            "start_time": m.get("start_time"),
+            "join_url": m.get("join_url"),
+        } for m in data.get("meetings", [])]
+        return _ok({"meetings": meetings, "next_page_token": data.get("next_page_token")})
+    except ZoomError as ze:
+        detail = (ze.payload or {}).get("message") or (ze.payload or {}).get("code") or (ze.payload or {}).get("text")
+        return _err(f"Zoom error {ze.status}: {detail}")
+
+@mcp.tool()
+async def create_meeting(
+    user_id_or_email: str,
+    topic: str,
+    start_time_iso: str,
+    duration_minutes: int = 30,
+    timezone: str | None = None,
+    settings: dict | None = None,
+) -> str:
+    """
+    Create a scheduled Zoom meeting for a given user.
+    Requires ISO 8601 start time format and supports custom settings.
+    """
+    try:
+        res = await zoom().create_meeting(
+            user_id_or_email=user_id_or_email,
+            topic=topic,
+            start_time_iso=start_time_iso,
+            duration_minutes=duration_minutes,
+            timezone=timezone,
+            settings=settings,
+        )
+        out = {
+            "meeting_id": res["id"],
+            "join_url": res["join_url"],
+            "start_url": res["start_url"],
+            "password": res.get("password"),
+        }
+        return _ok(out)
+    except ZoomError as ze:
+        detail = (ze.payload or {}).get("message") or (ze.payload or {}).get("code") or (ze.payload or {}).get("text")
+        return _err(f"Zoom error {ze.status}: {detail}")
+
+@mcp.tool()
+async def get_meeting(meeting_id: str | int) -> str:
+    """Fetch full details for a specific Zoom meeting."""
+    try:
+        res = await zoom().get_meeting(meeting_id)
+        return _ok(res)
+    except ZoomError as ze:
+        detail = (ze.payload or {}).get("message") or (ze.payload or {}).get("code") or (ze.payload or {}).get("text")
+        return _err(f"Zoom error {ze.status}: {detail}")
+
+@mcp.tool()
+async def update_meeting(
+    meeting_id: str | int | None = None,
+    patch_fields: dict | None = None
+) -> str:
+    """
+    Update selected fields for a meeting.
+    Example patch_fields:
+      {"topic":"New title","agenda":"Notes","settings":{"mute_upon_entry":true}}
+    """
+    if meeting_id in (None, ""):
+        return _err("Missing required parameter: meeting_id")
+    if not isinstance(patch_fields, dict) or not patch_fields:
+        return _err("Missing or invalid parameter: patch_fields (expected a non-empty object)")
+
+    try:
+        res = await zoom().update_meeting(meeting_id, patch_fields)
+        return _ok(res or {"updated": True, "meeting_id": meeting_id})
+    except ZoomError as ze:
+        detail = (ze.payload or {}).get("message") or (ze.payload or {}).get("code") or (ze.payload or {}).get("text")
+        return _err(f"Zoom error {ze.status}: {detail}")
+
+@mcp.tool()
+async def delete_meeting(meeting_id: str | int | None = None) -> str:
+    """Delete/cancel a Zoom meeting."""
+    if meeting_id in (None, ""):
+        return _err("Missing required parameter: meeting_id")
+
+    try:
+        await zoom().delete_meeting(meeting_id)
+        return _ok({"deleted": True, "meeting_id": meeting_id})
+    except ZoomError as ze:
+        detail = (ze.payload or {}).get("message") or (ze.payload or {}).get("code") or (ze.payload or {}).get("text")
+        return _err(f"Zoom error {ze.status}: {detail}")
+
+@mcp.tool()
+async def list_users(page_size: int = 30, status: str = "active") -> str:
+    """
+    List all users in the Zoom account.
+    Useful for finding target user IDs or emails.
+    """
+    try:
+        res = await zoom().list_users(page_size=page_size, status=status)
+        users = [
+            {
+                "id": u["id"],
+                "email": u.get("email"),
+                "first_name": u.get("first_name"),
+                "last_name": u.get("last_name")
+            }
+            for u in res.get("users", [])
+        ]
+        return _ok({"users": users, "next_page_token": res.get("next_page_token")})
+    except ZoomError as ze:
+        detail = (ze.payload or {}).get("message") or (ze.payload or {}).get("code") or (ze.payload or {}).get("text")
+        return _err(f"Zoom error {ze.status}: {detail}")
+
+
+# Start the MCP server
+if __name__ == "__main__":
+    mcp.run(transport="stdio")

mcp_servers/zoom/src/utils/schemas.py

@@ -0,0 +1,34 @@
+# src/utils/schemas.py
+from pydantic import BaseModel, Field, constr, conint
+from typing import Optional, Dict, Any, List
+
+EmailOrId = constr(strip_whitespace=True, min_length=1)
+
+class ListMeetingsInput(BaseModel):
+    user_id_or_email: EmailOrId = Field(..., description="Zoom user email or userId")
+    type: constr(strip_whitespace=True) = Field("upcoming", description='"upcoming" | "scheduled" | "past"')
+    page_size: conint(ge=1, le=100) = 10
+
+class MeetingItem(BaseModel):
+    id: int
+    topic: str
+    start_time: Optional[str] = None
+    join_url: Optional[str] = None
+
+class ListMeetingsOutput(BaseModel):
+    meetings: List[MeetingItem]
+    next_page_token: Optional[str] = None
+
+class CreateMeetingInput(BaseModel):
+    user_id_or_email: EmailOrId
+    topic: constr(strip_whitespace=True, min_length=1)
+    start_time_iso: constr(strip_whitespace=True, min_length=10)  # ISO8601
+    duration_minutes: conint(ge=1, le=1440) = 30
+    timezone: Optional[str] = None
+    settings: Optional[Dict[str, Any]] = None
+
+class CreateMeetingOutput(BaseModel):
+    meeting_id: int
+    join_url: str
+    start_url: str
+    password: Optional[str] = None