Merge pull request #553 from taylorwilsdon/issues/545

enh: expand get_form schema
This commit is contained in:
Taylor Wilsdon
2026-03-10 11:50:18 -04:00
committed by GitHub
2 changed files with 235 additions and 14 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[Dict[str, Any]]:
"""Extract valid option objects from Forms choice option objects.
Returns the full option dicts (preserving fields like ``isOther``,
``image``, ``goToAction``, and ``goToSectionId``) while filtering
out entries that lack a truthy ``value``.
"""
return [option for option in options if option.get("value")]
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,18 +192,24 @@ 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 = [
for i, item in enumerate(items, 1): _serialize_form_item(item, i) 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 = ( items_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"Item {item_index}")
item_type = serialized_item.get("type", "UNKNOWN")
required_text = " (Required)" if serialized_item.get("required") else ""
items_summary.append(
f" {item_index}. {item_title} [{item_type}]{required_text}"
)
items_summary_text = (
"\n".join(items_summary) if items_summary else " No items 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}:
- Title: "{title}" - Title: "{title}"
@@ -112,8 +218,10 @@ async def get_form(service, user_google_email: str, form_id: str) -> str:
- Form ID: {form_id} - Form ID: {form_id}
- Edit URL: {edit_url} - Edit URL: {edit_url}
- Responder URL: {responder_url} - Responder URL: {responder_url}
- Questions ({len(items)} total): - Items ({len(items)} total):
{questions_text}""" {items_summary_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,116 @@ 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"] == [{"value": "Red"}, {"value": "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"] == [{"value": "Never"}, {"value": "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