uv

core/cli_handler.py

@@ -0,0 +1,407 @@
+"""
+CLI Handler for Google Workspace MCP
+
+This module provides a command-line interface mode for directly invoking
+MCP tools without running the full server. Designed for use by coding agents
+(Codex, Claude Code) and command-line users.
+
+Usage:
+    workspace-mcp --cli                     # List available tools
+    workspace-mcp --cli list                # List available tools
+    workspace-mcp --cli <tool_name>         # Run tool (reads JSON args from stdin)
+    workspace-mcp --cli <tool_name> --args '{"key": "value"}'  # Run with inline args
+    workspace-mcp --cli <tool_name> --help  # Show tool details
+"""
+
+import asyncio
+import json
+import logging
+import sys
+from typing import Any, Dict, List, Optional
+
+from auth.oauth_config import set_transport_mode
+
+logger = logging.getLogger(__name__)
+
+
+def get_registered_tools(server) -> Dict[str, Any]:
+    """
+    Get all registered tools from the FastMCP server.
+
+    Args:
+        server: The FastMCP server instance
+
+    Returns:
+        Dictionary mapping tool names to their metadata
+    """
+    tools = {}
+
+    if hasattr(server, "_tool_manager") and hasattr(server._tool_manager, "_tools"):
+        tool_registry = server._tool_manager._tools
+        for name, tool in tool_registry.items():
+            tools[name] = {
+                "name": name,
+                "description": getattr(tool, "description", None)
+                or _extract_docstring(tool),
+                "parameters": _extract_parameters(tool),
+                "tool_obj": tool,
+            }
+
+    return tools
+
+
+def _extract_docstring(tool) -> Optional[str]:
+    """Extract the first meaningful line of a tool's docstring as its description."""
+    fn = getattr(tool, "fn", None) or tool
+    if fn and fn.__doc__:
+        # Get first non-empty line that's not just "Args:" etc.
+        for line in fn.__doc__.strip().split("\n"):
+            line = line.strip()
+            # Skip empty lines and common section headers
+            if line and not line.startswith(
+                ("Args:", "Returns:", "Raises:", "Example", "Note:")
+            ):
+                return line
+    return None
+
+
+def _extract_parameters(tool) -> Dict[str, Any]:
+    """Extract parameter information from a tool."""
+    params = {}
+
+    # Try to get parameters from the tool's schema
+    if hasattr(tool, "parameters"):
+        schema = tool.parameters
+        if isinstance(schema, dict):
+            props = schema.get("properties", {})
+            required = set(schema.get("required", []))
+            for name, prop in props.items():
+                params[name] = {
+                    "type": prop.get("type", "any"),
+                    "description": prop.get("description", ""),
+                    "required": name in required,
+                    "default": prop.get("default"),
+                }
+
+    return params
+
+
+def list_tools(server, output_format: str = "text") -> str:
+    """
+    List all available tools.
+
+    Args:
+        server: The FastMCP server instance
+        output_format: Output format ("text" or "json")
+
+    Returns:
+        Formatted string listing all tools
+    """
+    tools = get_registered_tools(server)
+
+    if output_format == "json":
+        # Return JSON format for programmatic use
+        tool_list = []
+        for name, info in sorted(tools.items()):
+            tool_list.append(
+                {
+                    "name": name,
+                    "description": info["description"],
+                    "parameters": info["parameters"],
+                }
+            )
+        return json.dumps({"tools": tool_list}, indent=2)
+
+    # Text format for human reading
+    lines = [
+        f"Available tools ({len(tools)}):",
+        "",
+    ]
+
+    # Group tools by service
+    services = {}
+    for name, info in tools.items():
+        # Extract service prefix from tool name
+        prefix = name.split("_")[0] if "_" in name else "other"
+        if prefix not in services:
+            services[prefix] = []
+        services[prefix].append((name, info))
+
+    for service in sorted(services.keys()):
+        lines.append(f"  {service.upper()}:")
+        for name, info in sorted(services[service]):
+            desc = info["description"] or "(no description)"
+            # Get first line only and truncate
+            first_line = desc.split("\n")[0].strip()
+            if len(first_line) > 70:
+                first_line = first_line[:67] + "..."
+            lines.append(f"    {name}")
+            lines.append(f"      {first_line}")
+        lines.append("")
+
+    lines.append("Use --cli <tool_name> --help for detailed tool information")
+    lines.append("Use --cli <tool_name> --args '{...}' to run a tool")
+
+    return "\n".join(lines)
+
+
+def show_tool_help(server, tool_name: str) -> str:
+    """
+    Show detailed help for a specific tool.
+
+    Args:
+        server: The FastMCP server instance
+        tool_name: Name of the tool
+
+    Returns:
+        Formatted help string for the tool
+    """
+    tools = get_registered_tools(server)
+
+    if tool_name not in tools:
+        available = ", ".join(sorted(tools.keys())[:10])
+        return f"Error: Tool '{tool_name}' not found.\n\nAvailable tools include: {available}..."
+
+    tool_info = tools[tool_name]
+    tool_obj = tool_info["tool_obj"]
+
+    # Get full docstring
+    fn = getattr(tool_obj, "fn", None) or tool_obj
+    docstring = fn.__doc__ if fn and fn.__doc__ else "(no documentation)"
+
+    lines = [
+        f"Tool: {tool_name}",
+        "=" * (len(tool_name) + 6),
+        "",
+        docstring,
+        "",
+        "Parameters:",
+    ]
+
+    params = tool_info["parameters"]
+    if params:
+        for name, param_info in params.items():
+            req = "(required)" if param_info.get("required") else "(optional)"
+            param_type = param_info.get("type", "any")
+            desc = param_info.get("description", "")
+            default = param_info.get("default")
+
+            lines.append(f"  {name}: {param_type} {req}")
+            if desc:
+                lines.append(f"    {desc}")
+            if default is not None:
+                lines.append(f"    Default: {default}")
+    else:
+        lines.append("  (no parameters)")
+
+    lines.extend(
+        [
+            "",
+            "Example usage:",
+            f'  workspace-mcp --cli {tool_name} --args \'{{"param": "value"}}\'',
+            "",
+            "Or pipe JSON from stdin:",
+            f'  echo \'{{"param": "value"}}\' | workspace-mcp --cli {tool_name}',
+        ]
+    )
+
+    return "\n".join(lines)
+
+
+async def run_tool(server, tool_name: str, args: Dict[str, Any]) -> str:
+    """
+    Execute a tool with the provided arguments.
+
+    Args:
+        server: The FastMCP server instance
+        tool_name: Name of the tool to execute
+        args: Dictionary of arguments to pass to the tool
+
+    Returns:
+        Tool result as a string
+    """
+    tools = get_registered_tools(server)
+
+    if tool_name not in tools:
+        raise ValueError(f"Tool '{tool_name}' not found")
+
+    tool_info = tools[tool_name]
+    tool_obj = tool_info["tool_obj"]
+
+    # Get the actual function to call
+    fn = getattr(tool_obj, "fn", None)
+    if fn is None:
+        raise ValueError(f"Tool '{tool_name}' has no callable function")
+
+    logger.debug(f"[CLI] Executing tool: {tool_name} with args: {list(args.keys())}")
+
+    try:
+        # Call the tool function
+        if asyncio.iscoroutinefunction(fn):
+            result = await fn(**args)
+        else:
+            result = fn(**args)
+
+        # Convert result to string if needed
+        if isinstance(result, str):
+            return result
+        else:
+            return json.dumps(result, indent=2, default=str)
+
+    except TypeError as e:
+        # Provide helpful error for missing/invalid arguments
+        error_msg = str(e)
+        params = tool_info["parameters"]
+        required = [n for n, p in params.items() if p.get("required")]
+
+        return (
+            f"Error calling {tool_name}: {error_msg}\n\n"
+            f"Required parameters: {required}\n"
+            f"Provided parameters: {list(args.keys())}"
+        )
+    except Exception as e:
+        logger.error(f"[CLI] Error executing {tool_name}: {e}", exc_info=True)
+        return f"Error: {type(e).__name__}: {e}"
+
+
+def parse_cli_args(args: List[str]) -> Dict[str, Any]:
+    """
+    Parse CLI arguments for tool execution.
+
+    Args:
+        args: List of arguments after --cli
+
+    Returns:
+        Dictionary with parsed values:
+            - command: "list", "help", or "run"
+            - tool_name: Name of tool (if applicable)
+            - tool_args: Arguments for the tool (if applicable)
+            - output_format: "text" or "json"
+    """
+    result = {
+        "command": "list",
+        "tool_name": None,
+        "tool_args": {},
+        "output_format": "text",
+    }
+
+    if not args:
+        return result
+
+    i = 0
+    while i < len(args):
+        arg = args[i]
+
+        if arg in ("list", "-l", "--list"):
+            result["command"] = "list"
+            i += 1
+        elif arg in ("--json", "-j"):
+            result["output_format"] = "json"
+            i += 1
+        elif arg in ("help", "--help", "-h"):
+            # Help command - if tool_name already set, show help for that tool
+            if result["tool_name"]:
+                result["command"] = "help"
+            else:
+                # Check if next arg is a tool name
+                if i + 1 < len(args) and not args[i + 1].startswith("-"):
+                    result["tool_name"] = args[i + 1]
+                    result["command"] = "help"
+                    i += 1
+                else:
+                    # No tool specified, show general help
+                    result["command"] = "list"
+            i += 1
+        elif arg in ("--args", "-a") and i + 1 < len(args):
+            # Parse inline JSON arguments
+            json_str = args[i + 1]
+            try:
+                result["tool_args"] = json.loads(json_str)
+            except json.JSONDecodeError as e:
+                # Provide helpful debug info
+                raise ValueError(
+                    f"Invalid JSON in --args: {e}\n"
+                    f"Received: {repr(json_str)}\n"
+                    f"Tip: Try using stdin instead: echo '<json>' | workspace-mcp --cli <tool>"
+                )
+            i += 2
+        elif not arg.startswith("-") and not result["tool_name"]:
+            # First non-flag argument is the tool name
+            result["tool_name"] = arg
+            result["command"] = "run"
+            i += 1
+        else:
+            i += 1
+
+    return result
+
+
+def read_stdin_args() -> Dict[str, Any]:
+    """
+    Read JSON arguments from stdin if available.
+
+    Returns:
+        Dictionary of arguments or empty dict if stdin is a TTY or no data is provided.
+    """
+    if sys.stdin.isatty():
+        logger.debug("[CLI] stdin is a TTY; no JSON args will be read from stdin")
+        return {}
+
+    try:
+        stdin_data = sys.stdin.read().strip()
+        if stdin_data:
+            return json.loads(stdin_data)
+    except json.JSONDecodeError as e:
+        raise ValueError(f"Invalid JSON from stdin: {e}")
+
+    return {}
+
+
+async def handle_cli_mode(server, cli_args: List[str]) -> int:
+    """
+    Main entry point for CLI mode.
+
+    Args:
+        server: The FastMCP server instance
+        cli_args: Arguments passed after --cli
+
+    Returns:
+        Exit code (0 for success, 1 for error)
+    """
+    # Set transport mode to "stdio" so OAuth callback server starts when needed
+    # This is required for authentication flow when no cached credentials exist
+    set_transport_mode("stdio")
+
+    try:
+        parsed = parse_cli_args(cli_args)
+
+        if parsed["command"] == "list":
+            output = list_tools(server, parsed["output_format"])
+            print(output)
+            return 0
+
+        if parsed["command"] == "help":
+            output = show_tool_help(server, parsed["tool_name"])
+            print(output)
+            return 0
+
+        if parsed["command"] == "run":
+            # Merge stdin args with inline args (inline takes precedence)
+            args = read_stdin_args()
+            args.update(parsed["tool_args"])
+
+            result = await run_tool(server, parsed["tool_name"], args)
+            print(result)
+            return 0
+
+        # Unknown command
+        print(f"Unknown command: {parsed['command']}")
+        return 1
+
+    except ValueError as e:
+        print(f"Error: {e}", file=sys.stderr)
+        return 1
+    except Exception as e:
+        logger.error(f"[CLI] Unexpected error: {e}", exc_info=True)
+        print(f"Error: {e}", file=sys.stderr)
+        return 1

core/server.py

@@ -13,6 +13,7 @@ from fastmcp.server.auth.providers.google import GoogleProvider
 
 from auth.oauth21_session_store import get_oauth21_session_store, set_auth_provider
 from auth.google_auth import handle_auth_callback, start_auth_flow, check_client_secrets
+from auth.oauth_config import is_oauth21_enabled, is_external_oauth21_provider
 from auth.mcp_session_middleware import MCPSessionMiddleware
 from auth.oauth_responses import (
     create_error_response,
@@ -358,15 +359,17 @@ def configure_server_for_http():
                     base_url=config.get_oauth_base_url(),
                     redirect_path=config.redirect_path,
                     required_scopes=required_scopes,
+                    resource_server_url=config.get_oauth_base_url(),
                 )
-                # Disable protocol-level auth, expect bearer tokens in tool calls
-                server.auth = None
-                logger.info(
-                    "OAuth 2.1 enabled with EXTERNAL provider mode - protocol-level auth disabled"
-                )
+                server.auth = provider
+
+                logger.info("OAuth 2.1 enabled with EXTERNAL provider mode")
                 logger.info(
                     "Expecting Authorization bearer tokens in tool call headers"
                 )
+                logger.info(
+                    "Protected resource metadata points to Google's authorization server"
+                )
             else:
                 # Standard OAuth 2.1 mode: use FastMCP's GoogleProvider
                 provider = GoogleProvider(
@@ -384,6 +387,18 @@ def configure_server_for_http():
                     "OAuth 2.1 enabled using FastMCP GoogleProvider with protocol-level auth"
                 )
 
+                # Explicitly mount well-known routes from the OAuth provider
+                # These should be auto-mounted but we ensure they're available
+                try:
+                    well_known_routes = provider.get_well_known_routes()
+                    for route in well_known_routes:
+                        logger.info(f"Mounting OAuth well-known route: {route.path}")
+                        server.custom_route(route.path, methods=list(route.methods))(
+                            route.endpoint
+                        )
+                except Exception as e:
+                    logger.warning(f"Could not mount well-known routes: {e}")
+
             # Always set auth provider for token validation in middleware
             set_auth_provider(provider)
             _auth_provider = provider
@@ -518,9 +533,9 @@ async def start_google_auth(
     """
     Manually initiate Google OAuth authentication flow.
 
-    NOTE: This tool should typically NOT be called directly. The authentication system
-    automatically handles credential checks and prompts for authentication when needed.
-    Only use this tool if:
+    NOTE: This is a legacy OAuth 2.0 tool and is disabled when OAuth 2.1 is enabled.
+    The authentication system automatically handles credential checks and prompts for
+    authentication when needed. Only use this tool if:
     1. You need to re-authenticate with different credentials
     2. You want to proactively authenticate before using other tools
     3. The automatic authentication flow failed and you need to retry
@@ -528,6 +543,19 @@ async def start_google_auth(
     In most cases, simply try calling the Google Workspace tool you need - it will
     automatically handle authentication if required.
     """
+    if is_oauth21_enabled():
+        if is_external_oauth21_provider():
+            return (
+                "start_google_auth is disabled when OAuth 2.1 is enabled. "
+                "Provide a valid OAuth 2.1 bearer token in the Authorization header "
+                "and retry the original tool."
+            )
+        return (
+            "start_google_auth is disabled when OAuth 2.1 is enabled. "
+            "Authenticate through your MCP client's OAuth 2.1 flow and retry the "
+            "original tool."
+        )
+
     if not user_google_email:
         raise ValueError("user_google_email must be provided.")
 

core/tool_registry.py

@@ -8,6 +8,9 @@ based on tier configuration, replacing direct @server.tool() decorators.
 import logging
 from typing import Set, Optional, Callable
 
+from auth.oauth_config import is_oauth21_enabled
+from auth.scopes import is_read_only_mode, get_all_read_only_scopes
+
 logger = logging.getLogger(__name__)
 
 # Global registry of enabled tools
@@ -79,7 +82,8 @@ def wrap_server_tool_method(server):
 def filter_server_tools(server):
     """Remove disabled tools from the server after registration."""
     enabled_tools = get_enabled_tools()
-    if enabled_tools is None:
+    oauth21_enabled = is_oauth21_enabled()
+    if enabled_tools is None and not oauth21_enabled:
         return
 
     tools_removed = 0
@@ -90,16 +94,56 @@ def filter_server_tools(server):
         if hasattr(tool_manager, "_tools"):
             tool_registry = tool_manager._tools
 
-            tools_to_remove = []
-            for tool_name in list(tool_registry.keys()):
-                if not is_tool_enabled(tool_name):
-                    tools_to_remove.append(tool_name)
+            read_only_mode = is_read_only_mode()
+            allowed_scopes = set(get_all_read_only_scopes()) if read_only_mode else None
+
+            tools_to_remove = set()
+
+            # 1. Tier filtering
+            if enabled_tools is not None:
+                for tool_name in list(tool_registry.keys()):
+                    if not is_tool_enabled(tool_name):
+                        tools_to_remove.add(tool_name)
+
+            # 2. OAuth 2.1 filtering
+            if oauth21_enabled and "start_google_auth" in tool_registry:
+                tools_to_remove.add("start_google_auth")
+                logger.info("OAuth 2.1 enabled: disabling start_google_auth tool")
+
+            # 3. Read-only mode filtering
+            if read_only_mode:
+                for tool_name in list(tool_registry.keys()):
+                    if tool_name in tools_to_remove:
+                        continue
+
+                    tool_func = tool_registry[tool_name]
+                    # Check if tool has required scopes attached (from @require_google_service)
+                    # Note: FastMCP wraps functions in Tool objects, so we need to check .fn if available
+                    func_to_check = tool_func
+                    if hasattr(tool_func, "fn"):
+                        func_to_check = tool_func.fn
+
+                    required_scopes = getattr(
+                        func_to_check, "_required_google_scopes", []
+                    )
+
+                    if required_scopes:
+                        # If ANY required scope is not in the allowed read-only scopes, disable the tool
+                        if not all(
+                            scope in allowed_scopes for scope in required_scopes
+                        ):
+                            logger.info(
+                                f"Read-only mode: Disabling tool '{tool_name}' (requires write scopes: {required_scopes})"
+                            )
+                            tools_to_remove.add(tool_name)
 
             for tool_name in tools_to_remove:
                 del tool_registry[tool_name]
                 tools_removed += 1
 
     if tools_removed > 0:
+        enabled_count = len(enabled_tools) if enabled_tools is not None else "all"
+        mode = "Read-Only" if is_read_only_mode() else "Full"
         logger.info(
-            f"Tool tier filtering: removed {tools_removed} tools, {len(enabled_tools)} enabled"
+            f"Tool filtering: removed {tools_removed} tools, {enabled_count} enabled. Mode: {mode}"
         )

core/tool_tiers.yaml

@@ -6,11 +6,15 @@ gmail:
     - send_gmail_message
 
   extended:
+    - get_gmail_attachment_content
     - get_gmail_thread_content
     - modify_gmail_message_labels
     - list_gmail_labels
     - manage_gmail_label
     - draft_gmail_message
+    - list_gmail_filters
+    - create_gmail_filter
+    - delete_gmail_filter
 
   complete:
     - get_gmail_threads_content_batch
@@ -34,6 +38,7 @@ drive:
     - get_drive_shareable_link
   extended:
     - list_drive_items
+    - copy_drive_file
     - update_drive_file
     - update_drive_permission
     - remove_drive_permission
@@ -63,6 +68,7 @@ calendar:
     - modify_event
   extended:
     - delete_event
+    - query_freebusy
   complete: []
 
 docs:
@@ -141,6 +147,7 @@ forms:
   complete:
     - set_publish_settings
     - get_form_response
+    - batch_update_form
 
 slides:
   core:
@@ -180,6 +187,26 @@ tasks:
     - move_task
     - clear_completed_tasks
 
+contacts:
+  core:
+    - search_contacts
+    - get_contact
+    - list_contacts
+    - create_contact
+  extended:
+    - update_contact
+    - delete_contact
+    - list_contact_groups
+    - get_contact_group
+  complete:
+    - batch_create_contacts
+    - batch_update_contacts
+    - batch_delete_contacts
+    - create_contact_group
+    - update_contact_group
+    - delete_contact_group
+    - modify_contact_group_members
+
 search:
   core:
     - search_custom
@@ -187,3 +214,25 @@ search:
     - search_custom_siterestrict
   complete:
     - get_search_engine_info
+
+appscript:
+  core:
+    - list_script_projects
+    - get_script_project
+    - get_script_content
+    - create_script_project
+    - update_script_content
+    - run_script_function
+    - generate_trigger_code
+  extended:
+    - create_deployment
+    - list_deployments
+    - update_deployment
+    - delete_deployment
+    - delete_script_project
+    - list_versions
+    - create_version
+    - get_version
+    - list_script_processes
+    - get_script_metrics
+  complete: []

core/utils.py

@@ -12,6 +12,7 @@ from typing import List, Optional
 from googleapiclient.errors import HttpError
 from .api_enablement import get_api_enablement_message
 from auth.google_auth import GoogleAuthenticationError
+from auth.oauth_config import is_oauth21_enabled, is_external_oauth21_provider
 
 logger = logging.getLogger(__name__)
 
@@ -314,10 +315,26 @@ def handle_http_errors(
                             )
                     elif error.resp.status in [401, 403]:
                         # Authentication/authorization errors
+                        if is_oauth21_enabled():
+                            if is_external_oauth21_provider():
+                                auth_hint = (
+                                    "LLM: Ask the user to provide a valid OAuth 2.1 "
+                                    "bearer token in the Authorization header and retry."
+                                )
+                            else:
+                                auth_hint = (
+                                    "LLM: Ask the user to authenticate via their MCP "
+                                    "client's OAuth 2.1 flow and retry."
+                                )
+                        else:
+                            auth_hint = (
+                                "LLM: Try 'start_google_auth' with the user's email "
+                                "and the appropriate service_name."
+                            )
                         message = (
                             f"API error in {tool_name}: {error}. "
                             f"You might need to re-authenticate for user '{user_google_email}'. "
-                            f"LLM: Try 'start_google_auth' with the user's email and the appropriate service_name."
+                            f"{auth_hint}"
                         )
                     else:
                         # Other HTTP errors (400 Bad Request, etc.) - don't suggest re-auth
@@ -336,6 +353,10 @@ def handle_http_errors(
                     logger.exception(message)
                     raise Exception(message) from e
 
+        # Propagate _required_google_scopes if present (for tool filtering)
+        if hasattr(func, "_required_google_scopes"):
+            wrapper._required_google_scopes = func._required_google_scopes
+
         return wrapper
 
     return decorator