MCP Google Workspace Server

Instructions

Implementation Reference

• src/mcp_google_suite/server.py:339-354 (handler)

  MCP tool handler that extracts arguments (document_id, content) and calls DocsService.update_document_content

  async def _handle_docs_update_content(
      self, context: GoogleWorkspaceContext, arguments: dict
  ) -> Dict[str, Any]:
      """Handle docs update content requests."""
      document_id = arguments.get("document_id")
      content = arguments.get("content")
  
      if not document_id or content is None:
          raise ValueError("Both document_id and content are required")
  
      logger.debug(f"Updating document - ID: {document_id}, Content length: {len(content)}")
      result = await context.docs.update_document_content(
          document_id=document_id, content=content
      )
      logger.debug("Document content updated successfully")
      return result

• src/mcp_google_suite/server.py:108-119 (schema)

  Input schema definition for the docs_update_content tool, specifying document_id and content as required string parameters

  types.Tool(
      name="docs_update_content",
      description="Update the content of a Google Doc",
      inputSchema={
          "type": "object",
          "properties": {
              "document_id": {"type": "string", "description": "ID of the document"},
              "content": {"type": "string", "description": "New content"},
          },
          "required": ["document_id", "content"],
      },
  ),

• src/mcp_google_suite/docs/service.py:39-53 (helper)

  Helper method in DocsService that performs the actual Google Docs API batchUpdate to insert new content at index 1 (effectively replacing content)

  async def update_document_content(self, document_id: str, content: str) -> Dict[str, Any]:
      """Update the content of a Google Doc."""
      try:
          service = await self.get_service()
          requests = [{"insertText": {"location": {"index": 1}, "text": content}}]
  
          result = await asyncio.to_thread(
              service.documents()
              .batchUpdate(documentId=document_id, body={"requests": requests})
              .execute
          )
  
          return {"success": True, "result": result}
      except HttpError as error:
          return {"success": False, **self.handle_error(error)}