Merge pull request #390 from ugoano/feature/forms-batch-update

feat(forms): Add batch_update_form tool for Google Forms API

Author: taylorwilsdon

README.md

@@ -818,6 +818,7 @@ cp .env.oauth21 .env
 | `set_publish_settings` | Complete | Configure form settings |
 | `get_form_response` | Complete | Get individual responses |
 | `list_form_responses` | Extended | List all responses with pagination |
+| `batch_update_form` | Complete | Apply batch updates (questions, settings) |
 
 </td>
 <td width="50%" valign="top">

README_NEW.md

@@ -141,7 +141,7 @@ export OAUTHLIB_INSECURE_TRANSPORT=1  # Development only
 
 **Comments:** `read_presentation_comments`, `create_presentation_comment`, `reply_to_presentation_comment`, `resolve_presentation_comment`
 
-### Google Forms (5 tools)
+### Google Forms (6 tools)
 
 | Tool | Tier | Description |
 |------|------|-------------|
@@ -150,6 +150,7 @@ export OAUTHLIB_INSECURE_TRANSPORT=1  # Development only
 | `list_form_responses` | Extended | List responses with pagination |
 | `set_publish_settings` | Complete | Configure template and authentication settings |
 | `get_form_response` | Complete | Get individual response details |
+| `batch_update_form` | Complete | Execute batch updates to forms (questions, items, settings) |
 
 ### Google Tasks (12 tools)
 

core/tool_tiers.yaml

@@ -108,6 +108,7 @@ forms:
   complete:
     - set_publish_settings
     - get_form_response
+    - batch_update_form
 
 slides:
   core:

gforms/forms_tools.py

@@ -6,7 +6,7 @@
 
 import logging
 import asyncio
-from typing import Optional, Dict, Any
+from typing import List, Optional, Dict, Any
 
 
 from auth.service_decorator import require_google_service
@@ -283,3 +283,97 @@ async def list_form_responses(
         f"Successfully retrieved {len(responses)} responses for {user_google_email}. Form ID: {form_id}"
     )
     return result
+
+
+# Internal implementation function for testing
+async def _batch_update_form_impl(
+    service: Any,
+    form_id: str,
+    requests: List[Dict[str, Any]],
+) -> str:
+    """Internal implementation for batch_update_form.
+
+    Applies batch updates to a Google Form using the Forms API batchUpdate method.
+
+    Args:
+        service: Google Forms API service client.
+        form_id: The ID of the form to update.
+        requests: List of update request dictionaries.
+
+    Returns:
+        Formatted string with batch update results.
+    """
+    body = {"requests": requests}
+
+    result = await asyncio.to_thread(
+        service.forms().batchUpdate(formId=form_id, body=body).execute
+    )
+
+    replies = result.get("replies", [])
+
+    confirmation_message = f"""Batch Update Completed:
+- Form ID: {form_id}
+- URL: https://docs.google.com/forms/d/{form_id}/edit
+- Requests Applied: {len(requests)}
+- Replies Received: {len(replies)}"""
+
+    if replies:
+        confirmation_message += "\n\nUpdate Results:"
+        for i, reply in enumerate(replies, 1):
+            if "createItem" in reply:
+                item_id = reply["createItem"].get("itemId", "Unknown")
+                question_ids = reply["createItem"].get("questionId", [])
+                question_info = (
+                    f" (Question IDs: {', '.join(question_ids)})"
+                    if question_ids
+                    else ""
+                )
+                confirmation_message += (
+                    f"\n  Request {i}: Created item {item_id}{question_info}"
+                )
+            else:
+                confirmation_message += f"\n  Request {i}: Operation completed"
+
+    return confirmation_message
+
+
+@server.tool()
+@handle_http_errors("batch_update_form", service_type="forms")
+@require_google_service("forms", "forms")
+async def batch_update_form(
+    service,
+    user_google_email: str,
+    form_id: str,
+    requests: List[Dict[str, Any]],
+) -> str:
+    """
+    Apply batch updates to a Google Form.
+
+    Supports adding, updating, and deleting form items, as well as updating
+    form metadata and settings. This is the primary method for modifying form
+    content after creation.
+
+    Args:
+        user_google_email (str): The user's Google email address. Required.
+        form_id (str): The ID of the form to update.
+        requests (List[Dict[str, Any]]): List of update requests to apply.
+            Supported request types:
+            - createItem: Add a new question or content item
+            - updateItem: Modify an existing item
+            - deleteItem: Remove an item
+            - moveItem: Reorder an item
+            - updateFormInfo: Update form title/description
+            - updateSettings: Modify form settings (e.g., quiz mode)
+
+    Returns:
+        str: Details about the batch update operation results.
+    """
+    logger.info(
+        f"[batch_update_form] Invoked. Email: '{user_google_email}', "
+        f"Form ID: '{form_id}', Requests: {len(requests)}"
+    )
+
+    result = await _batch_update_form_impl(service, form_id, requests)
+
+    logger.info(f"Batch update completed successfully for {user_google_email}")
+    return result

tests/gforms/__init__.py

@@ -0,0 +1,231 @@
+"""
+Unit tests for Google Forms MCP tools
+
+Tests the batch_update_form tool with mocked API responses
+"""
+
+import pytest
+from unittest.mock import Mock
+import sys
+import os
+
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../..")))
+
+# Import the internal implementation function (not the decorated one)
+from gforms.forms_tools import _batch_update_form_impl
+
+
+@pytest.mark.asyncio
+async def test_batch_update_form_multiple_requests():
+    """Test batch update with multiple requests returns formatted results"""
+    mock_service = Mock()
+    mock_response = {
+        "replies": [
+            {"createItem": {"itemId": "item001", "questionId": ["q001"]}},
+            {"createItem": {"itemId": "item002", "questionId": ["q002"]}},
+        ],
+        "writeControl": {"requiredRevisionId": "rev123"},
+    }
+
+    mock_service.forms().batchUpdate().execute.return_value = mock_response
+
+    requests = [
+        {
+            "createItem": {
+                "item": {
+                    "title": "What is your name?",
+                    "questionItem": {
+                        "question": {"textQuestion": {"paragraph": False}}
+                    },
+                },
+                "location": {"index": 0},
+            }
+        },
+        {
+            "createItem": {
+                "item": {
+                    "title": "What is your email?",
+                    "questionItem": {
+                        "question": {"textQuestion": {"paragraph": False}}
+                    },
+                },
+                "location": {"index": 1},
+            }
+        },
+    ]
+
+    result = await _batch_update_form_impl(
+        service=mock_service,
+        form_id="test_form_123",
+        requests=requests,
+    )
+
+    assert "Batch Update Completed" in result
+    assert "test_form_123" in result
+    assert "Requests Applied: 2" in result
+    assert "Replies Received: 2" in result
+    assert "item001" in result
+    assert "item002" in result
+
+
+@pytest.mark.asyncio
+async def test_batch_update_form_single_request():
+    """Test batch update with a single request"""
+    mock_service = Mock()
+    mock_response = {
+        "replies": [
+            {"createItem": {"itemId": "item001", "questionId": ["q001"]}},
+        ],
+    }
+
+    mock_service.forms().batchUpdate().execute.return_value = mock_response
+
+    requests = [
+        {
+            "createItem": {
+                "item": {
+                    "title": "Favourite colour?",
+                    "questionItem": {
+                        "question": {
+                            "choiceQuestion": {
+                                "type": "RADIO",
+                                "options": [
+                                    {"value": "Red"},
+                                    {"value": "Blue"},
+                                ],
+                            }
+                        }
+                    },
+                },
+                "location": {"index": 0},
+            }
+        },
+    ]
+
+    result = await _batch_update_form_impl(
+        service=mock_service,
+        form_id="single_form_456",
+        requests=requests,
+    )
+
+    assert "single_form_456" in result
+    assert "Requests Applied: 1" in result
+    assert "Replies Received: 1" in result
+
+
+@pytest.mark.asyncio
+async def test_batch_update_form_empty_replies():
+    """Test batch update when API returns no replies"""
+    mock_service = Mock()
+    mock_response = {
+        "replies": [],
+    }
+
+    mock_service.forms().batchUpdate().execute.return_value = mock_response
+
+    requests = [
+        {
+            "updateFormInfo": {
+                "info": {"description": "Updated description"},
+                "updateMask": "description",
+            }
+        },
+    ]
+
+    result = await _batch_update_form_impl(
+        service=mock_service,
+        form_id="info_form_789",
+        requests=requests,
+    )
+
+    assert "info_form_789" in result
+    assert "Requests Applied: 1" in result
+    assert "Replies Received: 0" in result
+
+
+@pytest.mark.asyncio
+async def test_batch_update_form_no_replies_key():
+    """Test batch update when API response lacks replies key"""
+    mock_service = Mock()
+    mock_response = {}
+
+    mock_service.forms().batchUpdate().execute.return_value = mock_response
+
+    requests = [
+        {
+            "updateSettings": {
+                "settings": {"quizSettings": {"isQuiz": True}},
+                "updateMask": "quizSettings.isQuiz",
+            }
+        },
+    ]
+
+    result = await _batch_update_form_impl(
+        service=mock_service,
+        form_id="quiz_form_000",
+        requests=requests,
+    )
+
+    assert "quiz_form_000" in result
+    assert "Requests Applied: 1" in result
+    assert "Replies Received: 0" in result
+
+
+@pytest.mark.asyncio
+async def test_batch_update_form_url_in_response():
+    """Test that the edit URL is included in the response"""
+    mock_service = Mock()
+    mock_response = {
+        "replies": [{}],
+    }
+
+    mock_service.forms().batchUpdate().execute.return_value = mock_response
+
+    requests = [
+        {"updateFormInfo": {"info": {"title": "New Title"}, "updateMask": "title"}}
+    ]
+
+    result = await _batch_update_form_impl(
+        service=mock_service,
+        form_id="url_form_abc",
+        requests=requests,
+    )
+
+    assert "https://docs.google.com/forms/d/url_form_abc/edit" in result
+
+
+@pytest.mark.asyncio
+async def test_batch_update_form_mixed_reply_types():
+    """Test batch update with createItem replies containing different fields"""
+    mock_service = Mock()
+    mock_response = {
+        "replies": [
+            {"createItem": {"itemId": "item_a", "questionId": ["qa"]}},
+            {},
+            {"createItem": {"itemId": "item_c"}},
+        ],
+    }
+
+    mock_service.forms().batchUpdate().execute.return_value = mock_response
+
+    requests = [
+        {"createItem": {"item": {"title": "Q1"}, "location": {"index": 0}}},
+        {
+            "updateFormInfo": {
+                "info": {"description": "Desc"},
+                "updateMask": "description",
+            }
+        },
+        {"createItem": {"item": {"title": "Q2"}, "location": {"index": 1}}},
+    ]
+
+    result = await _batch_update_form_impl(
+        service=mock_service,
+        form_id="mixed_form_xyz",
+        requests=requests,
+    )
+
+    assert "Requests Applied: 3" in result
+    assert "Replies Received: 3" in result
+    assert "item_a" in result
+    assert "item_c" in result