This commit is contained in:
Taylor Wilsdon
2026-03-09 12:20:04 -04:00
parent 6a2633984a
commit ea078623b3
2 changed files with 236 additions and 13 deletions

View File

@@ -6,6 +6,7 @@ This module provides MCP tools for interacting with Google Forms API.
import logging import logging
import asyncio import asyncio
import json
from typing import List, Optional, Dict, Any from typing import List, Optional, Dict, Any
@@ -16,6 +17,105 @@ from core.utils import handle_http_errors
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
def _extract_option_values(options: List[Dict[str, Any]]) -> List[str]:
"""Extract non-empty option values from Forms choice option objects."""
values: List[str] = []
for option in options:
value = option.get("value")
if value:
values.append(value)
return values
def _get_question_type(question: Dict[str, Any]) -> str:
"""Infer a stable question/item type label from a Forms question payload."""
choice_question = question.get("choiceQuestion")
if choice_question:
return choice_question.get("type", "CHOICE")
text_question = question.get("textQuestion")
if text_question:
return "PARAGRAPH" if text_question.get("paragraph") else "TEXT"
if "rowQuestion" in question:
return "GRID_ROW"
if "scaleQuestion" in question:
return "SCALE"
if "dateQuestion" in question:
return "DATE"
if "timeQuestion" in question:
return "TIME"
if "fileUploadQuestion" in question:
return "FILE_UPLOAD"
if "ratingQuestion" in question:
return "RATING"
return "QUESTION"
def _serialize_form_item(item: Dict[str, Any], index: int) -> Dict[str, Any]:
"""Serialize a Forms item with the key metadata agents need for edits."""
serialized_item: Dict[str, Any] = {
"index": index,
"itemId": item.get("itemId"),
"title": item.get("title", f"Question {index}"),
}
if item.get("description"):
serialized_item["description"] = item["description"]
if "questionItem" in item:
question = item.get("questionItem", {}).get("question", {})
serialized_item["type"] = _get_question_type(question)
serialized_item["required"] = question.get("required", False)
question_id = question.get("questionId")
if question_id:
serialized_item["questionId"] = question_id
choice_question = question.get("choiceQuestion")
if choice_question:
serialized_item["options"] = _extract_option_values(
choice_question.get("options", [])
)
return serialized_item
if "questionGroupItem" in item:
question_group = item.get("questionGroupItem", {})
columns = _extract_option_values(
question_group.get("grid", {}).get("columns", {}).get("options", [])
)
rows = []
for question in question_group.get("questions", []):
row: Dict[str, Any] = {
"title": question.get("rowQuestion", {}).get("title", "")
}
row_question_id = question.get("questionId")
if row_question_id:
row["questionId"] = row_question_id
row["required"] = question.get("required", False)
rows.append(row)
serialized_item["type"] = "GRID"
serialized_item["grid"] = {"rows": rows, "columns": columns}
return serialized_item
if "pageBreakItem" in item:
serialized_item["type"] = "PAGE_BREAK"
elif "textItem" in item:
serialized_item["type"] = "TEXT_ITEM"
elif "imageItem" in item:
serialized_item["type"] = "IMAGE"
elif "videoItem" in item:
serialized_item["type"] = "VIDEO"
else:
serialized_item["type"] = "UNKNOWN"
return serialized_item
@server.tool() @server.tool()
@handle_http_errors("create_form", service_type="forms") @handle_http_errors("create_form", service_type="forms")
@require_google_service("forms", "forms") @require_google_service("forms", "forms")
@@ -92,17 +192,23 @@ async def get_form(service, user_google_email: str, form_id: str) -> str:
) )
items = form.get("items", []) items = form.get("items", [])
questions_summary = [] serialized_items = [_serialize_form_item(item, i) for i, item in enumerate(items, 1)]
for i, item in enumerate(items, 1):
item_title = item.get("title", f"Question {i}")
item_type = (
item.get("questionItem", {}).get("question", {}).get("required", False)
)
required_text = " (Required)" if item_type else ""
questions_summary.append(f" {i}. {item_title}{required_text}")
questions_text = ( questions_summary = []
"\n".join(questions_summary) if questions_summary else " No questions found" for serialized_item in serialized_items:
item_index = serialized_item["index"]
item_title = serialized_item.get("title", f"Question {item_index}")
item_type = serialized_item.get("type", "UNKNOWN")
required_text = " (Required)" if serialized_item.get("required") else ""
questions_summary.append(
f" {item_index}. {item_title} [{item_type}]{required_text}"
)
questions_text = "\n".join(questions_summary) if questions_summary else " No questions found"
items_text = (
json.dumps(serialized_items, indent=2)
if serialized_items
else "[]"
) )
result = f"""Form Details for {user_google_email}: result = f"""Form Details for {user_google_email}:
@@ -113,7 +219,9 @@ async def get_form(service, user_google_email: str, form_id: str) -> str:
- Edit URL: {edit_url} - Edit URL: {edit_url}
- Responder URL: {responder_url} - Responder URL: {responder_url}
- Questions ({len(items)} total): - Questions ({len(items)} total):
{questions_text}""" {questions_text}
- Items (structured):
{items_text}"""
logger.info(f"Successfully retrieved form for {user_google_email}. ID: {form_id}") logger.info(f"Successfully retrieved form for {user_google_email}. ID: {form_id}")
return result return result

View File

@@ -11,8 +11,8 @@ import os
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../.."))) sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../..")))
# Import the internal implementation function (not the decorated one) # Import internal implementation functions (not decorated tool wrappers)
from gforms.forms_tools import _batch_update_form_impl from gforms.forms_tools import _batch_update_form_impl, _serialize_form_item, get_form
@pytest.mark.asyncio @pytest.mark.asyncio
@@ -229,3 +229,118 @@ async def test_batch_update_form_mixed_reply_types():
assert "Replies Received: 3" in result assert "Replies Received: 3" in result
assert "item_a" in result assert "item_a" in result
assert "item_c" in result assert "item_c" in result
def test_serialize_form_item_choice_question_includes_ids_and_options():
"""Choice question items should expose questionId/options/type metadata."""
item = {
"itemId": "item_123",
"title": "Favorite color?",
"questionItem": {
"question": {
"questionId": "q_123",
"required": True,
"choiceQuestion": {
"type": "RADIO",
"options": [{"value": "Red"}, {"value": "Blue"}],
},
}
},
}
serialized = _serialize_form_item(item, 1)
assert serialized["index"] == 1
assert serialized["itemId"] == "item_123"
assert serialized["type"] == "RADIO"
assert serialized["questionId"] == "q_123"
assert serialized["required"] is True
assert serialized["options"] == ["Red", "Blue"]
def test_serialize_form_item_grid_includes_row_and_column_structure():
"""Grid question groups should expose row labels/IDs and column options."""
item = {
"itemId": "grid_item_1",
"title": "Weekly chores",
"questionGroupItem": {
"questions": [
{
"questionId": "row_q1",
"required": True,
"rowQuestion": {"title": "Laundry"},
},
{
"questionId": "row_q2",
"required": False,
"rowQuestion": {"title": "Dishes"},
},
],
"grid": {
"columns": {"options": [{"value": "Never"}, {"value": "Often"}]}
},
},
}
serialized = _serialize_form_item(item, 2)
assert serialized["index"] == 2
assert serialized["type"] == "GRID"
assert serialized["grid"]["columns"] == ["Never", "Often"]
assert serialized["grid"]["rows"] == [
{"title": "Laundry", "questionId": "row_q1", "required": True},
{"title": "Dishes", "questionId": "row_q2", "required": False},
]
@pytest.mark.asyncio
async def test_get_form_returns_structured_item_metadata():
"""get_form should include question IDs, options, and grid structure."""
mock_service = Mock()
mock_service.forms().get().execute.return_value = {
"formId": "form_1",
"info": {"title": "Survey", "description": "Test survey"},
"items": [
{
"itemId": "item_1",
"title": "Favorite fruit?",
"questionItem": {
"question": {
"questionId": "q_1",
"required": True,
"choiceQuestion": {
"type": "RADIO",
"options": [{"value": "Apple"}, {"value": "Banana"}],
},
}
},
},
{
"itemId": "item_2",
"title": "Household chores",
"questionGroupItem": {
"questions": [
{
"questionId": "row_1",
"required": True,
"rowQuestion": {"title": "Laundry"},
}
],
"grid": {"columns": {"options": [{"value": "Never"}]}},
},
},
],
}
# Bypass decorators and call the core implementation directly.
result = await get_form.__wrapped__.__wrapped__(
mock_service, "user@example.com", "form_1"
)
assert "- Items (structured):" in result
assert '"questionId": "q_1"' in result
assert '"options": [' in result
assert '"Apple"' in result
assert '"type": "GRID"' in result
assert '"columns": [' in result
assert '"rows": [' in result